U Jc; @sddlZddlZddlZddlZddlZddlZddlZddlZddlmZddl m Z ddl m Z m Z ddlmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZddlZddlm Z!ddl"m m#m$m%Z%ddl&m'Z'ddl(m'm)Z*ddl+m,Z,ddlm-Z-ddl.m/Z/m0Z0m1Z1dd l"m2Z2dd l3m4Z4m5Z5dd l6m7Z7dd l8m9Z9m:Z:m;Z;dd lm?Z?m@Z@mAZAmBZBmCZCmDZDmEZEmFZFmGZGmHZHddlImJZJmKZKddlLmMZMmNZNmOZOmPZPmQZQmRZRmSZSddlTmUZUmVZVmWZWmXZXmYZYddlZm[Z[m\Z\m]Z]ddl^m_Z_m`Z`maZambZbdZczddldmeZemfZfWnegk rNdZcYnXdZheiedsddZhehr~ddljmkZkmlZlmmZmddddddd d!d"d#d$d%d&g Znd'Zoeod(e\d(Zpeqd)ZrGd*dde Zse Gd+ddZte Gd,ddZuGd-dde ZvGd.d%d%e ZwGd/dde Zxe Gd0d d Zye Gd1d!d!eyZze Gd2d"d"eyZ{e Gd3d#d#eyZ|exj}ezexj~e{exje|iZGd4d$d$e ZeeVd5fZGd6d7d7e ZGd8d9d9ZGd:d;d;ZesjeXjesjeXjesjeXjiZGdd?ZdJej'jeeej'j=eefd@dAdBZej'jeej'j=efdCdDdEZej'jeeej'j=fdCdFdGZeedHdId&ZdS)KN)contextmanager) dataclass)Enumauto)AnyCallableDequeDict GeneratorIterableIteratorListMapping NamedTupleOptionalSetTupleUnioncast)Variable) ProcessGroup)Shard ShardedTensorinit_from_local_shards)_CHECKPOINT_PREFIX)LOW_PRECISION_HOOKS default_hooks)_get_default_group)_replace_by_prefix_sync_params_and_buffers _to_kwargs) Parameter) _broadcast_pos_dim_tensor_states%_broadcast_processed_optim_state_dict_flatten_optim_state_dict_get_param_id_to_param'_get_param_id_to_param_from_optim_input_get_param_to_param_id'_get_param_to_param_id_from_optim_input_optim_state_dict_process_pos_dim_tensor_state_rekey_sharded_optim_state_dict)_ext_chunk_tensor"_ext_pre_load_state_dict_transform)_apply_to_modules_apply_to_tensors_contains_batchnorm _free_storage_is_fsdp_flattened#_override_batchnorm_mixed_precisionp_assert) FlatParameterFlatParamHandle HandleConfigHandleShardingStrategyHandleTrainingState) FLAT_PARAM FPW_MODULEFlattenParamsWrapper)ParamExecOrderWrapPolicy _or_policy_recursive_wrap_wrap_batchnorm_individuallyT) deferred_initfakeFZfx) TracingConfig_init_execution_info _patch_tracerFullyShardedDataParallelShardingStrategyMixedPrecision CPUOffloadBackwardPrefetch StateDictTypeStateDictConfigFullStateDictConfigLocalStateDictConfigShardedStateDictConfigOptimStateKeyTypeTrainingState_clean_tensor_name_fsdp_wrapped_module.ic@s"eZdZdZeZeZeZdS)rHa This specifies the sharding strategy to be used for distributed training by :class:`FullyShardedDataParallel`. FULL_SHARD: Parameters, gradients, and optimizer states are sharded. For the parameters, this algorithm all-gathers before the forward, reshards after the forward, all-gathers before the backward computation, and reshards after the backward computation. The gradients are synchronized and sharded via reduce-scatter after the backward computation. The sharded optimizer states are updated locally. SHARD_GRAD_OP: Gradients and optimizer states are sharded during computation, and additionally parameters are sharded outside computation. For the parameters, this algorithm all-gathers before the forward, does not reshard after the forward, and only reshards after the backward computation. The gradients are synchronized and sharded via reduce-scatter after the backward computation. The sharded optimizer states are updated locally. Inside ``no_sync()``, the parameters are not resharded after the backward computation. NO_SHARD: Parameters, gradients, and optimizer states are not sharded but instead replicated across ranks, similar to PyTorch's ``DistributedDataParallel`` API. The gradients are synchronized via all-reduce after the backward computation. The unsharded optimizer states are updated locally. HYBRID_SHARD(future support): Apply ``FULL_SHARD`` intra-node and ``NO_SHARD`` inter-node. N)__name__ __module__ __qualname____doc__r FULL_SHARD SHARD_GRAD_OPNO_SHARDr]r]V/tmp/pip-unpacked-wheel-gikjz4vx/torch/distributed/fsdp/fully_sharded_data_parallel.pyrHsc@sXeZdZUdZdZeejed<dZ eejed<dZ eejed<dZ ee ed<dS)rIa A config to enable mixed precision training with FullyShardedDataParallel. This class can be constructed with several flags: ``param_dtype`` controls the precision of model parameters, inputs, and therefore the precision under which computation happens. After forward and backward passes, FSDP parameters point to full precision shards that are kept in memory. Full precision parameters are always checkpointed. ``reduce_dtype`` controls the precision under which gradient reduction would occur, which can potentially be different than ``param_dtype`` for use cases such as communication efficiency. ``buffer_dtype`` controls the precision that buffers are cast to. Note that buffers are unsharded and are cast in the first forward pass, and remain in their reduced precision state even after forward/backward passes. However, when taking checkpoints with ``state_dict``, buffers are checkpointed in their full precision (and then restored back to to their reduced precision) as expected. Note that this checkpoint support is currently limited to ``StateDictType.FULL_STATE_DICT``. ``keep_low_precision_grads``: Whether to upcast gradients back to the full parameter precision after backwards or not. This can be disabled to keep the gradients in the lower precision, which can potentially save memory if custom Optimizers are able to perform parameter updates effectively with lower precision grads. .. note:: In ``summon_full_params``, parameters are summoned in full precision but buffers are not. .. note:: Parameters and buffers are checkpointed in full precision. For buffers, this is only guaranteed to work for ``StateDictType.FULL_STATE_DICT``. .. note:: This API is experimental and subject to change. .. note:: Specification of reduced precision types must be explicit, in that if, for example, ``param_dtype`` is not specified, it will not be cast by FSDP. Thus, a config such as ``MixedPrecision(reduce_dtype=torch.float16)`` will not cast buffers or parameters. Note that if a ``MixedPrecision`` config is specified without a ``reduce_dtype``, gradient communication would occur in the `param_dtype` precision, if given, otherwise, in the original parameter precision. N param_dtype reduce_dtype buffer_dtypeFkeep_low_precision_grads) rVrWrXrYr_rtorchdtype__annotations__r`rarbboolr]r]r]r^rIs +c@seZdZUdZdZeed<dS)rJa CPU offloading config. Currently, only parameter and gradient CPU offload are supported. offload_params: Offloading parameters to CPUs when these parameters are not used for computation on GPUs. This implicitly enables gradient offloading to CPUs in order for parameters and gradients to be on the same device to work with optimizer. Foffload_paramsN)rVrWrXrYrgrfrer]r]r]r^rJs  c@seZdZdZeZeZdS)rKa  Specify where to prefetch next layer's full parameters during backward pass. BACKWARD_PRE: prefetch right before current layer's backward computation starts, this approach will increase backward communication and computation overalpping and potentialy improve training performance, but it may increase the peak memory usage as the prefetched full parameters will be kept in the GPU memory until next layer's backward computation is done. BACKWARD_POST: prefetch right after current layer's backward computation finishes, this approach will not increase peak memory as prefetching happens after current layer's full parameters are freed. It could potentially improve backward communication and computation overlapping as it avoids all_gather and reduce_scatter are blocked each other in the single NCCL stream. However, based on our experiments, for some models, the backward post backward hook fire order is not always the reversed forward computation order, so this approach may prefetch full parameters for layers ahead of next layer, this 'ahead' all_gather could delay next layer's all_gather in the single NCCL stream and cause the next layer's computation delay. So it may cause some performance regession for some models. N)rVrWrXrYr BACKWARD_PRE BACKWARD_POSTr]r]r]r^rKsc@s.eZdZdZeZeZeZeZeZ dS)rRa Simple enum to indicate what state FSDP is in. Used for asserting to make sure APIs are called in the correct state. ..note:: ``BACKWARD_PRE`` and ``BACKWARD_POST`` states are used to ensure we receives backward hooks in the correct order. It is used to catch unexpected order of hooks being called (likely due to our hook registration logic or autograd engine logic changes). N) rVrWrXrYrIDLEFORWARDrhriSUMMON_FULL_PARAMSr]r]r]r^rRs  c@s"eZdZdZeZeZeZdS)rLa This enum indicates that which type of ``state_dict`` the FSDP module is currently processing (returning or loading). The default value is FULL_STATE_DICT to comply the PyTorch convention. ..note:: FSDP currently supports three types of ``state_dict``: 1. ``state_dict/load_state_dict`: this pair of APIs return and load the non-sharded, unflattened parameters. The semantics is the same as using DDP. 2. ``_local_state_dict/_load_local_state_dict``: this pair of APIs return and load local sharded, flattened parameters. The values returned by ``_local_state_dict`` can be directly used by FSDP and is only meaningful to FSDP (because parameters are flattened). Note that these APIs are meant for use via the :func:`state_dict_type` context manager as follows: >>> # xdoctest: +SKIP("undefined variables") >>> with fsdp.state_dict_type(StateDictType.LOCAL_STATE_DICT): ... state = fsdp.state_dict() # loads local state dict 3. ``_sharded_state_dict/_load_sharded_state_dict``: this pair of APIs return and load sharded, unflattened parameters. The ``state_dict`` return by ``sharded_state_dict`` can be used by all other parallel schemes (resharding may be required). N)rVrWrXrYrFULL_STATE_DICTLOCAL_STATE_DICTSHARDED_STATE_DICTr]r]r]r^rLsc@seZdZdZdS)rMa ``StateDictConfig`` is the base class for all state_dict configuration classes. Users should instantiate a child version (i.e. ``FullStateDictConfig``) in order to configure settings for the particular type of ``state_dict`` implementation FSDP will use. N)rVrWrXrYr]r]r]r^rM6sc@s*eZdZUdZdZeed<dZeed<dS)rNa ``FullStateDictConfig`` is a config class meant to be used with ``StateDictType.FULL_STATE_DICT``. Currently, it accepts two parameters, ``offload_to_cpu`` and ``rank0_only`` which can be configured to offload the full ``state_dict`` to CPU and to materialize the ``state_dict`` on rank 0 only. When used, it is recommended to enable both of these flags together to optimize memory savings when taking checkpoints. Note that this config class is meant for user via the :func:`state_dict_type` context manager as follows: >>> # xdoctest: +SKIP("undefined variables") >>> fsdp = FSDP(model, auto_wrap_policy=...) >>> cfg = FullStateDictConfig(offload_to_cpu=True, rank0_only=True) >>> with FullyShardedDataParallel.state_dict_type(fsdp, StateDictType.FULL_STATE_DICT, cfg): >>> state = fsdp.state_dict() >>> # state will be empty on non rank 0 and contain CPU tensors on rank 0. >>> # To reload checkpoint for inference, finetuning, transfer learning, etc: >>> model = model_fn() # Initialize model on CPU in preparation for wrapping with FSDP >>> if dist.get_rank() == 0: >>> # Load checkpoint only on rank 0 to avoid memory redundancy >>> state_dict = torch.load("my_checkpoint.pt") >>> model.load_state_dict(state_dict) >>> # All ranks initialize FSDP module as usual. ``sync_module_states`` argument >>> # communicates loaded checkpoint states from rank 0 to rest of the world. >>> fsdp = FSDP(model, device_id=torch.cuda.current_device(), auto_wrap_policy=..., sync_module_states=True) >>> # After this point, all ranks have FSDP model with loaded checkpoint. Foffload_to_cpu rank0_onlyN)rVrWrXrYrprfrerqr]r]r]r^rN@s  c@s eZdZdS)rONrVrWrXr]r]r]r^rO_sc@s eZdZdS)rPNrrr]r]r]r^rPcsc@seZdZeZeZdS)rQN)rVrWrXr PARAM_NAMEPARAM_IDr]r]r]r^rQms.c@s"eZdZdZeZeZeZdS)_ExecOrderWarnStatusz/Used internally for execution order validation.N)rVrWrXrYrNONEWARNINGWARNEDr]r]r]r^ruxsruc@seZdZdZejeeddddZdejdddd Z e e e d d d Z e e e d d dZ e eddddZe eeddddZe eddddZe eeedfdddZeedfe e edddZe e e eddd Zd!d"ZdS)#_ExecOrderDataae This contains the data structures to track the execution order. We track the pre-forward order on the *first* iteration for forward prefetching (which thus assumes static graph) and the post-forward order on *every* iteration for backward prefetching (which thus does not assume static graph but may be provide an incorrect order). N) debug_levelbackward_prefetch_limitforward_prefetch_limitreturncCspg|_i|_g|_i|_d|_||_||_|tjj tjj fk|_ d|_ d|_ g|_i|_i|_d|_tj|_dS)NTr)handles_pre_forward_order"handles_to_pre_forward_order_indexhandles_post_forward_order#handles_to_post_forward_order_index is_first_iter_backward_prefetch_limit_forward_prefetch_limitdist DebugLevelINFOZDETAIL_checking_order process_group world_size all_handleshandle_to_handle_index"flat_param_to_prefixed_param_namescurrent_order_indexrurv warn_status)selfrzr{r|r]r]r^__init__s z_ExecOrderData.__init__rG) fsdp_rootrr}cCsv||_||_||_||D]0}|jD]$}t|j}|j|||j |<q.q$t t t t tft||_dS)z Initializes the data structures needed for checking the forward order. This should be called after a root FSDP instance has been set during lazy initialization. N)rranksizer fsdp_modules_handleslenrappendrrr r6r str _get_param_to_unflat_param_namesr)rrr fsdp_modulehandleindexr]r]r^inits      z_ExecOrderData.initcurrent_handles_keyr}cCs^|j|d}|dkrdS|d}g}t|jD](}|dkr@qZ||j||d8}q0|S)z Returns a :class:`list` of the handles keys of the handles to backward prefetch given the current handles key. If there are no valid handles keys to prefetch, then this returns an empty :class:`list`. Nr"r)rgetrangerrrrrZ current_indexZ target_indextarget_handles_keys_r]r]r^ get_handles_to_backward_prefetchs  z/_ExecOrderData.get_handles_to_backward_prefetchcCsd|j|d}|dkrdS|d}g}t|jD].}|t|jkrFq`||j||d7}q0|S)z Returns a :class:`list` of the handles keys of the handles to forward prefetch given the current handles key. If there are no valid handles keys to prefetch, then this returns an empty :class:`list`. Nr")rrrrrr~rrr]r]r^get_handles_to_forward_prefetchs  z._ExecOrderData.get_handles_to_forward_prefetchhandlesr}cCsB|sdSt|}||jkrdSt|j}||j|<|j|dS)a Records ``handles`` in the post-forward order, where ``handles`` should be a group of handles used in the same module's forward. If ``handles`` is empty, then it is omitted. Unlike :meth:`record_pre_forward`, this records the order *every* iteration with the expectation that the recorded order is reset in :meth:`next_iter`. N)tuplerrrr)rr handles_keyrr]r]r^record_post_forwards    z"_ExecOrderData.record_post_forward)r is_trainingr}cCsT|sdSt|}||||jr,||jkr0dSt|j}||j|<|j|dS)ab Records ``handles`` in the pre-forward order on the first iteration, where ``handles`` should be a group of handles used in the same module's forward. If ``handles`` is empty, then it is omitted. On the first iteration, this checks the execution order across ranks. See :meth:`_check_order` for details. N)r _check_orderrrrr~r)rrrrrr]r]r^record_pre_forward s    z!_ExecOrderData.record_pre_forward)rrr}c sB|sdS|jrd}||}|dj}tdd|Dtj|d}tj|jf|tjgf|}t j ||j dt fddt|jDd D]>\\}} \} } | | krt|d |d | d | d | d qtj|jf|tj|f|}t j ||j dt fddt|jDd D]T\\}} \} } | | kr.|| }|| }t|d |d|d| d| q.n|jr>|jtjkrdSd}|jt|jkrd}n,|j|j}||kr||}d|d}|dk r0||}|r|nd}td|jd||tj|_|jd7_dS)a Checks the forward execution order as long as ``is_training`` is ``True`` since checking in eval mode is not supported. - On the first iteration, this uses all-gathers to check that all ranks are all-gathering the same handles and hence ``FlatParameter`` s, raising an error if not. - On subsequent iterations, if the distributed debug level is at least INFO, then this checks that each rank is locally consistent with its own forward order from the first iteration, issuing a warning if not. This issues a warning on the first deviating iteration and stops warning thereafter. Nz#Forward order differs across ranks:rcss|]}|dk VqdSNr]).0rr]r]r^ ;sz._ExecOrderData._check_order..rddevicegroupc3s|]}||fVqdSrr]rr)world_num_valid_indicesr]r^rIsz rank z is all-gathering z parameters while rank z parametersc3s*|]"}|||dfVqdS)r"Nr]r)num_valid_indices world_indicesr]r^r]s z! is all-gathering parameters for z while rank zfExpected to not all-gather any more parameters in the forward but trying to all-gather parameters for zExpected to all-gather for z) but trying to all-gather parameters for z/a newly-added parameter since construction timez?Forward order differs from that of the first iteration on rank zD. Collectives are unchecked and may give incorrect results or hang. r")r_get_handle_indicesrsumrcZint32zerosrtensorr_all_gather_baser itertools combinationsr RuntimeError_get_names_from_handle_indicesrrrurxrrr~_get_names_from_handleswarningswarnrrw)rrr msg_prefixZ local_indicesrZ tensor_kwargsZlocal_num_valid_indicesZr1Zn1Zr2Zn2i1i2Zr1_param_namesZr2_param_namesZexpected_handles_keyZexpected_param_names param_namesZ msg_suffixr])rrrr^r#s                z_ExecOrderData._check_order.rr}cCs<g}|D]*}||jkr"|dq||j|qt|S)z Returns the handle indices (i.e. indices into ``self.all_handles``) corresponding to the handles in ``handles_key``. An entry in the returned tuple is ``None`` if the handle is invalid. N)rrr)rrindicesrr]r]r^rs   z"_ExecOrderData._get_handle_indices)handle_indicesr}cCsRg}|D]D}|dks|dks|t|jkr,q|j|}|j}||j|q|S)z Returns a list of prefixed parameter names for each handle in ``handle_indices``. If a handle index is invalid, then its prefixed parameter names are omitted from the returned list. Nr)rr flat_paramrr)rrprefixed_param_namesrrrr]r]r^rs  z-_ExecOrderData._get_names_from_handle_indicescCs4g}|D]&}|j}||jkrq||j|q|S)z Returns a list of prefixed parameter names for each handle in ``handles_key``. If a handle is invalid, then its prefixed parameter names are omitted from the returned list. )rrr)rrrrrr]r]r^rs  z&_ExecOrderData._get_names_from_handlescCs>d|_|j|j|jr:d|_|jtjkr:tj |_dS)z Advances the internal data structures per iteration. This should be called in the post-backward callback since that marks the true end of an iteration. FrN) rrclearrrrrrurwrxrr]r]r^ next_iters   z_ExecOrderData.next_iter)rVrWrXrYrrintrrr _HandlesKeyr rrr7rrfrrrrrrrrrr]r]r]r^rys:  (   v    ryc@s^eZdZdZddddZejjddddZe ejjdd d Z e ejjdd d Z dS) _FreeEventQueuez This tracks all pending frees corresponding to inflight all-gathers. The queueing pattern is iterative enqueues with a single dequeue per iteration once the limit ``_max_num_inflight_all_gathers`` is reached. Nr}cCst|_d|_dS)Nr) collectionsdeque_queue_max_num_inflight_all_gathersrr]r]r^rs z_FreeEventQueue.__init__) free_eventr}cCs|j|dS)zEnqueues a free event.N)rr)rrr]r]r^enqueuesz_FreeEventQueue.enqueuecCst|j|jkr|SdS)z0Dequeues a single event if the limit is reached.N)rrr_dequeuerr]r]r^dequeue_if_neededsz!_FreeEventQueue.dequeue_if_neededcCs|jr|j}|SdS)z"Dequeues a free event if possible.N)rpopleft)reventr]r]r^rs z_FreeEventQueue._dequeue) rVrWrXrYrrccudaEventrrrrr]r]r]r^rs rcs eZdZdZdejeeeeee ee ee ee ee ejjee ejgdfeeeejfeeed fdd Zejee ejjeejddd Zejjeejjeeejjeefd d d Zejeed ddZeeefeeefddddZejeejddddZeeeejfeejdddZejee ejgdfeejddddZ ejeejeejdddZ!ejeejeejejdd d!Z"eje#ejdd"d#d$Z$ejeeje%ejdd%d&Z&eejdd'd(d)Z'e(dd*d+d,Z)e*e#e(dd-d.d/Z+e#e(e#edd0d1d2Z,e-ejd3d4d5Z.eed6fd7d8 Z/eed9d:d;Z0ed3dd?d@Z3e ejgdfddAfdBdC Z4ed3dDdEZ5ed3dFdGZ6ed3dHdIZ7ed3dJdKZ8ed3dLdMZ9ej:eeeeefdNdOdPZ;deejeeeej:feeeddRdSdTZe*e(dd*dYdZZ?dd3d[d\Z@dd3d]d^ZAeBdd_d`daZCeBe#eBd_dbdcZDeBeEdddedfZFe2eGjHdejeIeeJeKdgdhdiZLeedjdkdlZMe-e%eeeefd3dmdnZNeeefeeeefdodpdqZOeeefeeeefdodrdsZPe*eeefeeeefdodtduZQe2ejeeefeeeeefdvdwdxZRfdydzZSeeed{d|d}ZTdd3d~dZUeeed{ddZVeeefeddoddZWdd3ddZXeeefeddoddZYdd3ddZZeeefeddoddZ[e2ejeeefeeddvddZ\e2ejeddddZ]e^eefe_dfdd Z`e^eefe_dddZadeeeejbfdfee_dddZceeed{ddZde#e(ee ejedddZee#e(dd-ddZfe#e(ee ejeeedddZgddZhddZie2eGjHdeeeeeKdddZjeGjHdeeeedddZke*e#e(dddZle%eeejbfd3fdd Zme%eeejjfd3fdd Znee#e(edddZoe#e(dd-ddZpe*e(eddddZqejberdddZse(ddd„Ztdd3ddĄZue*dd3ddƄZvdd3ddȄZweexe#exfddɜdd˄ZyeHeKd3dd̈́Zze-e#ed3ddτZ{e*dee|efee|efddќddӄZ}e2ddՄZ~e2ed3ddׄZe2dejjejjeee#eeefe ejjfeeejeeefd؜ddڄZe2dejjejjeee#eeefe ejjfeejeeefdۜdd݄Ze2deeefejjeee#eeefe ejjfeejjeeefdޜddZe2deeefejjeee#eeefe ejjfeejjeeefdddZe2deeeefejjeee#eeefe ejjfeejjeeeeefdddZe2deeefeejjeee#eeefe ejjfeejjeeefdddZed3ddZed3ddZeedddZdd3ddZed3ddZed3ddZZS(rGa62 A wrapper for sharding Module parameters across data parallel workers. This is inspired by `Xu et al.`_ as well as the ZeRO Stage 3 from DeepSpeed_. FullyShardedDataParallel is commonly shortened to FSDP. .. _`Xu et al.`: https://arxiv.org/abs/2004.13336 .. _DeepSpeed: https://www.deepspeed.ai/ Example:: >>> # xdoctest: +SKIP("undefined variables") >>> import torch >>> from torch.distributed.fsdp import FullyShardedDataParallel as FSDP >>> torch.cuda.set_device(device_id) >>> sharded_module = FSDP(my_module) >>> optim = torch.optim.Adam(sharded_module.parameters(), lr=0.0001) >>> x = sharded_module(x, y=3, z=torch.Tensor([1])) >>> loss = x.sum() >>> loss.backward() >>> optim.step() .. warning:: The optimizer must be initialized *after* the module has been wrapped, since FSDP will shard parameters in-place and this will break any previously initialized optimizers. .. warning:: If the destination CUDA device has ID ``dev_id``, either (1) ``module`` should already be placed on that device, (2) the device should be set using ``torch.cuda.set_device(dev_id)``, or (3) ``dev_id`` should be passed into the ``device_id`` constructor argument. This FSDP instance's compute device will be that destination device. For (1) and (3), the FSDP initialization always occurs on GPU. For (2), the FSDP initialization happens on ``module`` 's current device, which may be CPU. .. warning:: FSDP currently does not support gradient accumulation outside ``no_sync()`` when using CPU offloading. Trying to do so yields incorrect results since FSDP will use the newly-reduced gradient instead of accumulating with any existing gradient. .. warning:: Changing the original parameter variable names after construction will lead to undefined behavior. .. warning:: Passing in `sync_module_states=True` flag requires module to be put on GPU, or to use ``device_id`` argument to specify a CUDA device that FSDP will move module to. This is because ``sync_module_states=True`` requires GPU communication. .. warning:: As of PyTorch 1.12, FSDP only offers limited support for shared parameters (for example, setting one ``Linear`` layer's weight to another's). In particular, modules that share parameters must be wrapped as part of the same FSDP unit. If enhanced shared parameter support is needed for your use case, please ping https://github.com/pytorch/pytorch/issues/77724 .. note:: Inputs into FSDP ``forward`` function will be moved to compute device (same device FSDP module is on) before running ``forward``, so user does not have to manually move inputs from CPU -> GPU. Args: module (nn.Module): module to be wrapped with FSDP. process_group (Optional[ProcessGroup]): process group for sharding sharding_strategy (Optional[ShardingStrategy]): Config sharding algorithm, different sharding algorithm has trade off between memory saving and communication overhead. ``FULL_SHARD`` will be chosen if sharding_strategy is not specified. cpu_offload (Optional[CPUOffload]): CPU offloading config. Currently, only parameter and gradient CPU offload is supported. It can be enabled via passing in ``cpu_offload=CPUOffload(offload_params=True)``. Note that this currently implicitly enables gradient offloading to CPU in order for params and grads to be on same device to work with optimizer. This API is subject to change. Default is ``None`` in which case there will be no offloading. auto_wrap_policy (Optional[Callable[[nn.Module, bool, int], bool]]): A callable specifying a policy to recursively wrap layers with FSDP. Note that this policy currently will only apply to child modules of the passed in module. The remainder modules are always wrapped in the returned FSDP root instance. ``size_based_auto_wrap_policy`` written in ``torch.distributed.fsdp.wrap`` is an example of ``auto_wrap_policy`` callable, this policy wraps layers with the number of parameters larger than 100M. ``transformer_auto_wrap_policy`` written in ``torch.distributed.fsdp.wrap`` is an example of ``auto_wrap_policy`` callable for transformer-like model architectures. Users can supply the customized ``auto_wrap_policy`` callable that should accept following arguments: ``module: nn.Module``, ``recurse: bool``, ``unwrapped_params: int``, and return a ``bool`` specifying whether the passed in ``module``` should be wrapped (if ``recurse=False``) or whether we should recurse down the subgraph of ``module`` children (if ``recurse=True``). Extra customized arguments could be added to the customized ``auto_wrap_policy`` callable as well. It is a good practice to print out the sharded model and check whether the sharded model is what the application wants and then adjust accordingly. Example:: >>> def custom_auto_wrap_policy( >>> module: nn.Module, >>> recurse: bool, >>> unwrapped_params: int, >>> # These are customizable for this policy function. >>> min_num_params: int = int(1e8), >>> ) -> bool: >>> return unwrapped_params >= min_num_params >>> # Configure a custom min_num_params >>> my_auto_wrap_policy = functools.partial(custom_auto_wrap_policy, min_num_params=1e5) backward_prefetch (Optional[BackwardPrefetch]): This is an experimental feature that is subject to change in the the near future. It allows users to enable two different backward_prefetch algorithms to help backward communication and computation overlapping. Pros and cons of each algorithm is explained in the class ``BackwardPrefetch``. mixed_precision (Optional[MixedPrecision]): A ``MixedPrecision`` instance describing the mixed precision training config to be used. ``MixedPrecision`` supports configuring parameter, buffer, and gradient communication dtype. Note that only floating point data is cast to the reduced precision. This allows users potential memory saving and training speedup while trading off accuracy during model training. If ``None``, no mixed precision is applied. Note that if ``mixed_precision`` is enabled for FSDP model that contains ``BatchNorm`` with ``auto_wrap_policy``, FSDP will take care to disable mixed precision for ``BatchNorm`` units by wrapping them separately in their own FSDP unit with ``mixed_precision=None``. This is done because several ``BatchNorm`` kernels do not implement reduced type support at the moment. If individually wrapping the model, users must take care to set ``mixed_precision=None`` for ``BatchNorm`` units. (Default: ``None``) ignored_modules (Optional[Iterable[torch.nn.Module]]): Modules whose own parameters and child modules' parameters and buffers are ignored by this instance. None of the modules directly in ``ignored_modules`` should be :class:`FullyShardedDataParallel` instances, and any child modules that are already-constructed :class:`FullyShardedDataParallel` instances will not be ignored if they are nested under this instance. This argument may be used to avoid sharding specific parameters at module granularity when using an ``auto_wrap_policy`` or if parameters' sharding is not managed by FSDP. (Default: ``None``) param_init_fn (Optional[Callable[[nn.Module], None]]): A ``Callable[torch.nn.Module] -> None`` that specifies how modules that are currently on the meta device should be initialized onto an actual device. Note that as of v1.12, we detect modules on the meta device via ``is_meta`` check and apply a default initialization that calls ``reset_parameters`` method on the passed in ``nn.Module`` if ``param_init_fn`` is not specified, otherwise we run ``param_init_fn`` to initialize the passed in ``nn.Module``. In particular, this means that if ``is_meta=True`` for any module parameters for modules that will be wrapped with FSDP and ``param_init_fn`` is not specified, we assume your module properly implements a ``reset_paramters()`` and will throw errors if not. Note that additionally, we offer support for modules initialized with torchdistX's (https://github.com/pytorch/torchdistX) ``deferred_init`` API. In this case, deferred modules would be initialized by a default initialization function that calls torchdistX's ``materialize_module``, or the passed in ``param_init_fn``, if it is not ``None``. The same ``Callable`` is applied to initialize all meta modules. Note that this initialization function is applied before doing any FSDP sharding logic. Example:: >>> # xdoctest: +SKIP("undefined variables") >>> module = MyModule(device="meta") >>> def my_init_fn(module): >>> # responsible for initializing a module, such as with reset_parameters >>> ... >>> fsdp_model = FSDP(module, param_init_fn=my_init_fn, auto_wrap_policy=size_based_auto_wrap_policy) >>> print(next(fsdp_model.parameters()).device) # current CUDA device >>> # With torchdistX >>> module = deferred_init.deferred_init(MyModule, device="cuda") >>> # Will initialize via deferred_init.materialize_module(). >>> fsdp_model = FSDP(module, auto_wrap_policy=size_based_auto_wrap_policy) device_id (Optional[Union[int, torch.device]]): An ``int`` or ``torch.device`` describing the CUDA device the FSDP module should be moved to determining where initialization such as sharding takes place. If this argument is not specified and ``module`` is on CPU, we issue a warning mentioning that this argument can be specified for faster initialization. If specified, resulting FSDP instances will reside on this device, including moving ignored modules' parameters if needed. Note that if ``device_id`` is specified but ``module`` is already on a different CUDA device, an error will be thrown. (Default: ``None``) sync_module_states (bool): If ``True``, each individually wrapped FSDP unit will broadcast module parameters from rank 0 to ensure they are the same across all ranks after initialization. This helps ensure model parameters are the same across ranks before starting training, but adds communication overhead to ``__init__``, as at least one broadcast is triggered per individually wrapped FSDP unit. This can also help load checkpoints taken by ``state_dict`` and to be loaded by ``load_state_dict`` in a memory efficient way. See documentation for :class:`FullStateDictConfig` for an example of this. (Default: ``False``) forward_prefetch (bool): If ``True``, then FSDP *explicitly* prefetches the next upcoming all-gather while executing in the forward pass. This may improve communication and computation overlap for CPU bound workloads. This should only be used for static graph models since the forward order is fixed based on the first iteration's execution. (Default: ``False``) limit_all_gathers (bool): If ``False``, then FSDP allows the CPU thread to schedule all-gathers without any extra synchronization. If ``True``, then FSDP explicitly synchronizes the CPU thread to prevent too many in-flight all-gathers. This ``bool`` only affects the sharded strategies that schedule all-gathers. Enabling this can help lower the number of CUDA malloc retries. NF modulersharding_strategy cpu_offloadauto_wrap_policybackward_prefetchmixed_precisionignored_modules param_init_fn device_idsync_module_statesforward_prefetchlimit_all_gatherscsdt|tr2|j||||||||| | | | | d dStjdt||||_ | ||j \}|_ | ||_ |dk r||t|j |dd}|||||| | | | | d }||||pt|_|j|_|j|_tj|_|pt|_||_| |_| |_d}d}|jdkrtj}|p&tj|_ |p4t!|_"i|_#|$|||%| }|&|| ||'||||(||||_)t*|+||}| r|,||t-t.|j |jj/|j"j0|j"j1|j"j2}t3|||j)||_4|5|g|_6g|_7|j4j8rh|j4j9}|j7:|j;|<||=|j|jj/rh|j;j>t>dkrht?|@t>dW5QRXd|_A|B|_C|D|_Ed|_Fi|_Gd|_Hi|_ItJ|_KtLM|_NtO|jN|||_Pi|_Qi|_Ri|_StTjU|_VtW|_X|Y|jZtTjU|j[tTj\|j]tTj^|j_i|_`|ja|jbdd tTjU|jctTj\|jdtTj^|jei|_f|g|jhtTjU|jitTj\|jjtTj^|jki|_ldS) Nrztorch.distributed.fsdpT)rrZ wrapper_clsrignored_paramsZonly_wrap_children) rrrrrrrrrrr"cpuF)Z with_module)m isinstancer>"_init_param_exec_order_wrap_policyrcZ_CZ_log_api_usage_oncesuperr_get_ignored_modules_ignored_modules_get_ignored_params_ignored_param_names_get_buffer_names _buffer_namesrG _auto_wraprrrrrrRrjtraining_staterJrrrrrHr\rZrrIr_buffer_name_to_orig_dtype_check_single_device_module_get_device_from_device_id_materialize_module_move_module_to_device_get_compute_devicecompute_devicelist_get_orig_params_sync_module_statesr8sharding_strategy_maprgr_r`rbr=rT_check_orig_params_flattenedrparams has_paramsrrr_register_param_handleshardrno_grad_flat_param_to_sync_gradients_get_default_comm_hook_communication_hook_get_default_comm_hook_state_communication_hook_state_hook_registered_ran_pre_backward_hook_is_root_streamsr_free_event_queuerZget_debug_levelZ _debug_levelry_exec_order_data_handles_prefetched_needs_pre_backward_unshard_needs_pre_forward_unshardrLrm_state_dict_typerN_state_dict_configZ_register_state_dict_hook_post_state_dict_hook_full_post_state_dict_hookrn_local_post_state_dict_hookro_sharded_post_state_dict_hook_post_state_dict_hook_fnZ"_register_load_state_dict_pre_hook_pre_load_state_dict_hook_full_pre_load_state_dict_hook_local_pre_load_state_dict_hook!_sharded_pre_load_state_dict_hook_pre_load_state_dict_hook_fnZ"register_load_state_dict_post_hook_post_load_state_dict_hook_full_post_load_state_dict_hook _local_post_load_state_dict_hook"_sharded_post_load_state_dict_hook_post_load_state_dict_hook_fn)rrrrrrrrrrrrrrrauto_wrap_kwargs fsdp_kwargsr{r|device_from_device_idZparams_to_flattenconfigr __class__r]r^rs                        z!FullyShardedDataParallel.__init__) root_modulerr}cCs|dkrtSd}z t|}Wn*tk rHt|dt|YnX|D]:}t|tjjsvt|dt|t|trNtdqNtdd|D}||krt d|| D](}t|trt |d st ||jq|S) a3 Checks that ``_ignored_modules`` is an iterable of ``nn.Module`` s without any FSDP instances, and returns the modules contained in their module subtrees as a :class:`set`. Nested FSDP instances are excluded, but their already-computed ignored modules are included. Nz>`ignored_modules` should be an iterable of `torch.nn.Module`s zbut got zbut got an iterable with z1`ignored_modules` should not include FSDP modulescss.|]&}|D]}t|ttfs|VqqdSr)modulesrrGr=)rrchildr]r]r^rs  z@FullyShardedDataParallel._get_ignored_modules..zTrying to ignore the top-level module passed into the FSDP constructor itself will result in all parameters being ignored and is not well-supported: r)set TypeErrortyperrcnnModulerG ValueErrorrrr6hasattrAssertionErrorupdater)rr5rrZignored_root_modulesrr submoduler]r]r^rs0      z-FullyShardedDataParallel._get_ignored_modules)r5rr}c Csdtdd|D}t|dd}t}|D]2}||}g}|D]} |t| q<||q(||fS)z Returns the parameters of the modules in ``ignored_modules``, excluding any :class:`FlatParameter` s, and their fully prefixed names, both as :class:`set` s. css(|] }|D]}t|s|VqqdSr) parametersr3)rmpr]r]r^rs  z?FullyShardedDataParallel._get_ignored_params..Fdedup_shared_params)r8rrrSr@) rr5rrparam_to_unflat_param_namesZignored_param_namesparamZunflat_param_namesZ clean_nameskr]r]r^rs  z,FullyShardedDataParallel._get_ignored_params)r5r}cCs>tjtttddd}ttddd}t}t||||S)z Returns the fully prefixed names of all buffers in the module hierarchy rooted at ``root_module`` as a class:`set`. )rprefix buffer_namescSs:t|ts6|jddD]\}}t||}||qdS)NFrecurse)rrG named_buffersrSadd)rrJrK buffer_namerZprefixed_buffer_namer]r]r^ module_fns  z=FullyShardedDataParallel._get_buffer_names..module_fn)rKcWs|Srr])rKargsr]r]r^ return_fnsz=FullyShardedDataParallel._get_buffer_names..return_fn)r;r<rrr8r/)rr5rQrSrKr]r]r^rs z*FullyShardedDataParallel._get_buffer_names)r/r0r}cCs|d}|d}|dk st|D]"\}}t|tr$td|dq$|d}|dk rt|rt|tjt t |gd}t d||d<t f||dS) a Recursively auto wraps the root module given by the key "module" in ``auto_wrap_kwargs`` with the arguments in ``auto_wrap_kwargs`` and ``fsdp_kwargs``. Precondition: ``auto_wrap_policy`` contains the arguments expected by ``_recursive_wrap()``, where ``auto_wrap_policy`` is not ``None``. ``fsdp_kwargs`` contains all FSDP arguments except ``module``. rrN Expected zB to NOT be FullyShardedDataParallel if using an `auto_wrap_policy`r)ZpoliciesaBoth mixed precision and an `auto_wrap_policy` were specified for FSDP, where the wrapped module has batch norm submodules. The batch norm submodules will be wrapped as separate FSDP instances with mixed precision disabled since some batch norm kernels do not support low precision.)r?Z named_modulesrrGr=r1r4 functoolspartialr?rArrr@)rr/r0rr5 module_namerrr]r]r^rs(   z#FullyShardedDataParallel._auto_wrap)rrr}cCs8tdd|||D}t|dkr4td|dS)a Raises an error if ``module`` has original parameters on multiple devices, ignoring the parameters in ``ignored_params``. Thus, after this method, the module must be either fully on the CPU or fully on a non-CPU device. css|] }|jVqdSrrrrHr]r]r^rszGFullyShardedDataParallel._check_single_device_module..r"z;FSDP only supports single device modules but got params on N)r8rrr)rrrZdevicesr]r]r^r s   z4FullyShardedDataParallel._check_single_device_module)rr}c Csp|dkr dSt|tjr|nt|}|tdkrltd|d|jdtjdtdtj}|S)z Nrz"FSDP got the argument `device_id` z on rank zJ, which does not have an explicit index. FSDP will use the current device z. If this is incorrect, please explicitly call `torch.cuda.set_device()` before FSDP initialization or pass in the explicit device index as the `device_id` argument.)rrcrrrrrcurrent_device)rrrr]r]r^r s z3FullyShardedDataParallel._get_device_from_device_id)rrr1r}c Cs tdd|D}| o4to4tdd|D}|s>|rp|dk rpt|sftd|dt|||n|r|ptj}|j |dz t | W5QRXWn<t k r}zt dt|d |W5d}~XYnXn|rtj|d d d dS) ae Materializes the wrapped module ``module`` in place if needed: either if the module has parameters that use meta device or are torchdistX fake tensors. This method uses ``param_init_fn`` to materialize the module if the function is not ``None`` and falls back to default behavior otherwise. For meta device, this moves the module to ``device_from_device_id`` if it is not ``None`` or the current device otherwise and calls ``reset_parameters()``, and for torchdistX fake tensors, this calls ``deferred_init.materialize_module()``. css|] }|jVqdSr)is_metarrDr]r]r^rKsz?FullyShardedDataParallel._materialize_module..css|]}t|VqdSr)rCZis_faker\r]r]r^rOsNrTz to be callable but got rXzIUnable to call `reset_parameters()` for module on meta device with error zE. Please ensure your module implements a `reset_parameters()` method.cSs t|t Sr)rrG)rIr]r]r^kz>FullyShardedDataParallel._materialize_module..)Zcheck_fn)anyrB_TORCHDISTX_AVAILcallabler=r:rcrrZZto_emptyrZreset_parameters BaseExceptionrrrrBZmaterialize_module)rrrr1Zis_meta_moduleZis_torchdistX_deferred_initZmaterialization_deviceer]r]r^r9sB   z,FullyShardedDataParallel._materialize_module)rrr1c Cstd}t|||d}|dkr(dS|dk r|j|kr||}|D]F}t|trL|jj rLt "|j D]}| tdqrW5QRXqLn|j|krt ddS)a Moves ``module`` depending on ``device_from_device_id`` and its current device. This includes moving ignored modules' parameters. - If ``device_from_device_id`` is not ``None``, then this moves ``module`` to the device. - If ``device_from_device_id`` is ``None``, then this does not move ``module`` but warns the user if it is on CPU. Precondition: ``_check_single_device_module()``. rNaAModule is put on CPU and will thus have flattening and sharding run on CPU, which is less efficient than on GPU. We recommend passing in `device_id` argument which will enable FSDP to put module on GPU device, module must also be on GPU device to work with `sync_module_states=True` flag which requires GPU communication.)rcrnextrtor6rrGrrgrrrrr)rrrr1 cpu_devicerHrArr]r]r^rns&        z/FullyShardedDataParallel._move_module_to_device)rrr1r}cCspt|||d}|dk r.|jjdkr.|j}ntdtj}|dk rl||krltd|jd|d||S)aI Determines and returns this FSDP instance's compute device. If the module is already on a non-CPU device, then the compute device is that non-CPU device. If the module is on CPU, then the compute device is the current device. Since this method should be called after materializing the module, any non-CPU device should not be meta device. For now, the compute device is always a CUDA GPU device with its explicit index. Precondition: ``_check_single_device_module()`` and ``_move_module_to_device()``. Nrz4Inconsistent compute device and `device_id` on rank : z vs ) rdrrr:rcrrZr=r)rrrr1rHrr]r]r^rsz,FullyShardedDataParallel._get_compute_device)rr r}cCsz|rtdd|Drtdg}|D]$}t|dds*d|_||q*|dd|Dt|j |t dd d S) a Synchronizes module states (i.e. parameters ``params`` and all not-yet-synced buffers) by broadcasting from rank 0 to all ranks. Precondition: ``sync_module_states == True`` and ``self.process_group`` has been set. css|]}|jtdkVqdS)rN)rrcrYr]r]r^rsz?FullyShardedDataParallel._sync_module_states..zModule has CPU parameters, but sync_module_states=True is specified.This only works for GPU module, please specify `device_id` argument or move module to GPU before init. _fsdp_syncedFTcss|]}|VqdSr)detachrYr]r]r^rsr)srcN) r_r=buffersgetattrrhrriextendrr_PARAM_BROADCAST_BUCKET_SIZE)rrr Z module_statesbufferr]r]r^rs   z,FullyShardedDataParallel._sync_module_statesccsF|}z$t|}||kr t|s |Vq Wntk r@YnXdS)z Returns an iterator over the original parameters in ``module``, ignoring the parameters in ``ignored_params`` and any ``FlatParameter`` s (which may be present due to nested FSDP wrapping). N)rBrdr3 StopIteration)rrrZ param_genrHr]r]r^rs  z)FullyShardedDataParallel._get_orig_params)rr}cCsF|D]8\}}||krt|std|d|d|jqdS)z Checks that all original parameters have been flattened and hence made invisible to ``named_parameters()``. This should be called as a sanity check after flattening the wrapped module's parameters. z Found an unflattened parameter: z;  N)named_parametersr3rrr4)rr param_namerHr]r]r^r s z5FullyShardedDataParallel._check_orig_params_flattened)rr}cCs||jkr|j|dS)z5Registers the parameter handle to this FSDP instance.N)rrrrr]r]r^r s z/FullyShardedDataParallel._register_param_handlerc Cs|sdS|jr$|j}|r$|d}tj|jd |D]}|}|pR|}q@W5QRX|rz|jd |jdtj|jd |D]}| | qW5QRXdS)aI Unshards the handles in ``handles``. If the handles are in :meth:`summon_full_params` and are using mixed precision, then they are forced to full precision. Postcondition: Each handle's ``FlatParameter`` 's data is the padded unsharded flattened parameter on the compute device. NFpre_all_gather all_gather) rrr synchronizercrstreamrZ pre_unshard wait_streamZunshardZ post_unshard)rrrZany_ran_pre_unshardrZran_pre_unshardr]r]r^_unshards"  z!FullyShardedDataParallel._unshard)rfree_unsharded_flat_paramsr}cCs|sdStt|t|kdt|dt|t||D]B\}}|||jrv|rvtj}||j || qms  z9FullyShardedDataParallel.fsdp_modules..)r6)rrr]rr^rZs z%FullyShardedDataParallel.fsdp_modules)fnr}c sb|jdk}|tj|jdddt|}W5QRX|r^|jr^||D] }|qP|S)a:Applies ``fn`` recursively to every submodule (as returned by ``.children()``) as well as self. Typical use includes initializing the parameters of a model (see also :ref:`nn-init-doc`). Compared to ``torch.nn.Module.apply``, this version additionally gathers the full parameters before applying ``fn``. It should not be called from within another ``summon_full_params`` context. Args: fn (:class:`Module` -> None): function to be applied to each submodule Returns: Module: self NFTrM writeback) r _assert_staterRrj_summon_full_paramsrapplyr_reset_lazy_init)rrZ uninitializedretrr3r]r^rss    zFullyShardedDataParallel.applycCs |jjdk S)z` Whether user explicitly enabled mixed precision for parameters or not. N)rr_rr]r]r^#_mixed_precision_enabled_for_paramssz.cast_fnN)rcTensorrr0)rrdrRrrr]rr^_cast_fp_inputs_to_dtypes  z1FullyShardedDataParallel._cast_fp_inputs_to_dtypeT)rrdmemorMr}cCs|dkrt}|D]}||k rDt|trD|rD|j||||dq||kr|||jddD]|\}}|dkrtqb|j|p|jd}||j kr|j |j |<t |r|r|j||d}n| r||jj}t|||qbqdS)aMove all buffers to the given *device* and *dtype*. If *device* is not given, then it will default to ``self.compute_device``, otherwise buffer will be moved to ``device``. In the case of nested FSDP instances, we will respect the child instance's ``compute_device`` configuration. If *dtype* is given, it must be a mapping of buffer name to buffer dtype, and this argument is currently only given to restore back to original buffer types during checkpoint. If *dtype* is not given, and we are in mixed precision training, the buffer will be cast to buffer_dtype, otherwise the buffer will not be cast. Args: device (torch.device, Optional): device to cast buffers to (defaults to compute_device) dtype: (Dict[str, torch.dtype], Optional): Mapping of buffer name to their dtype to cast to. memo (Set, Optional): set of modules that have already been processed recurse (bool, Optional): Whether to call _cast_buffers recursively on nested FSDP instances (default is True). N)rrdrrMFrLrXr)r8r6rrG _cast_buffersrOrNrerrrdrcrrrrasetattr)rrrdrrMrrbufr]r]r^rs&     z&FullyShardedDataParallel._cast_bufferscCs$d|_|jD]}t|dr |`q dS)zT Reset instance so :func:`_lazy_init` will run on the next forward. N _local_shard)rr r>r)rrDr]r]r^rs  z)FullyShardedDataParallel._reset_lazy_initcCs"|jdk rdStjs tdd|_|tj||j dd|j D]}| |qL|j ||jd}||D]}||k rz|jdks|jrtdd|_|j|_|j |_ |j|jkrd}|j|_|j|_|j|_|j|_|j D]}| |qqz|rtd|jd|jd dS) a Performs initialization lazily, typically right before the first forward pass. The laziness is needed to ensure that the parameter device/dtype and the FSDP hierarchy have finalized. This method's actual logic only runs on the root FSDP instance, which performs initialization for all non-root FSDP instances to avoid partial initialization. Nz(FSDP does not support CPU only executionTrLFzcNon-root FSDP instance's `_is_root` should not have been set yet or should have been set to `False`zLFound inconsistent `limit_all_gathers` values across FSDP instances on rank z!. Using the root FSDP's value of z for all instances.)rrcr is_availablerrrRrj _init_streamsrr_init_param_attributesrrrrr?rrrrrrrr)rrZinconsistent_limit_all_gathersrr]r]r^rs@        z#FullyShardedDataParallel._lazy_initcCsh|j}t|dr@|jjr<|jjtdksrrgrrrcr?dataZ pin_memory zeros_like _cpu_gradrrrr_Z _mp_shardr2uses_sharded_strategyrdrnumelrZ_full_param_paddedrZ_padded_unsharded_sizeZ_full_prec_full_param_padded_post_backward_called)rrrDZfull_param_dtyper]r]r^rMs`            z/FullyShardedDataParallel._init_param_attributescCsL|js ttjsttj|jd<tj|jd<tj|jd<dS)zInitializes CUDA streams for overlapping data transfer and computation. This should only be called on the root FSDP instance.rv post_backwardruN)rr?rcrrZStreamrrr]r]r^rs  z&FullyShardedDataParallel._init_streamscCs8|js dStj}|jd||jd|dS)z The root :class:`FullyShardedDataParallel` instance needs to synchronize with the default stream to ensure that the previous optimizer step is done. Nrvru)rrcrcurrent_streamrry)rrr]r]r^_wait_for_previous_optim_steps  z6FullyShardedDataParallel._wait_for_previous_optim_steprcCs4|sdS||}|D]}||d|j|<qdS)z Prefetches the next handles if needed (without synchronization). An empty handles key cannot prefetch. NT)_get_handles_to_prefetchrzr)rrZhandles_to_prefetchrr]r]r^_prefetch_handless   z*FullyShardedDataParallel._prefetch_handlescs|}tjtjtjf}t||kd|d|j}g}|tjkrTjtjksj|tjkrjtjkrfdd| |D}n(|tjkrj rfdd| |D}|S)ap Returns a :class:`list` of the handles keys to prefetch for the next module(s), where ``current_handles_key`` represents the current module. "Prefetching" refers to running the unshard logic early (without synchronization), and the "next" modules depend on the recorded execution order and the current training state. z!Prefetching is only supported in z but currently in cs,g|]$}j|drj|ds|qSF)rrrrZtarget_handles_keyrr]r^rszEFullyShardedDataParallel._get_handles_to_prefetch..cs,g|]$}j|drj|ds|qSr)rrrrrr]r^rs) _get_training_stater:rhrirkr5rrrKrrr)rrrZvalid_training_statesZeodrr]rr^rs<     z1FullyShardedDataParallel._get_handles_to_prefetchrcCsHtt|dkdtdd|D}tt|dkd|tt|S)z=Returns the training state of the handles in ``handles_key``.rzExpects a non-empty handles keycss|] }|jVqdSr)_training_staterrr]r]r^rsz?FullyShardedDataParallel._get_training_state..r"z'Expects uniform training state but got )r5rr8rditer)rrZtraining_statesr]r]r^rs z,FullyShardedDataParallel._get_training_state)rstate_dict_typestate_dict_configr}c csd}d}|dkrt|}t|D]}|dkr6|j}|dkrD|j}||jkrVtdt|t|jkrptdt|}|t|krtd|dt|||_||_q$z dVW5|dk st|dk stt|D]}||_||_qXdS)a A context manager to set the ``state_dict_type`` of all the descendant FSDP modules of the target module. The target module does not have to be a FSDP module. If the target module is a FSDP module, its ``state_dict_type`` will also be changed. .. note:: This API should be called for only the top-level (root) module. .. note:: This API enables users to transparently use the conventional ``state_dict`` API to take model checkpoints in cases where the root FSDP module is wrapped by another ``nn.Module``. For example, the following will ensure ``state_dict`` is called on all non-FSDP instances, while dispatching into `local_state_dict` implementation for FSDP: Example:: >>> # xdoctest: +SKIP("undefined variables") >>> model = DDP(FSDP(...)) >>> with FSDP.state_dict_type(model, StateDictType.LOCAL_STATE_DICT): >>> checkpoint = model.state_dict() Args: module (torch.nn.Module): Root module. state_dict_type (StateDictType): the desired ``state_dict_type`` to set. Nz0All FSDP module should the same state_dict_type.z@All FSDP modules should have the same type of state_dict_config.z#Expected state_dict_config of type but got )_state_dict_type_to_configrGrrrrr:r?)rrrZprev_state_dict_typeZprev_state_dict_configrAZexpected_state_dict_config_typer]r]r^r%s:"      z(FullyShardedDataParallel.state_dict_type)rWr}cCsF|tdd}|td}|r.|d}|tjdd}|S)NrU)replacer<checkpoint_wrapperr)rrWr]r]r^_convert_to_wrapped_module_namehs  z8FullyShardedDataParallel._convert_to_wrapped_module_nameccs<|jjD]*\}}||}||}|||fVq dSr)rTrZparameter_module_namesr)rrsrWfqnr]r]r^ _param_fqnsts    z$FullyShardedDataParallel._param_fqns) state_dictrJr}c Cst||td||tjg|r0|jjs4|St|jdrD|S|jj }t d}|j D]\}}}||}|}t |} || r|t| d}||kstd|d|d|d|d |d |jd ||jkr\t||d d s\z"||||<d ||_Wq\tk rT} z$td|d|dt| W5d} ~ XYq\Xq\|r|jD]V}|tjdd}||}||krqd||j |krd|| |||<qd|S)a) Hook that runs after model.state_dict() is called before returning result to user. For FSDP, we may have to clone the tensors in state_dict as params go back to sharded version after _summon_full_params ends, and also remove "_fsdp_wrapped_module" prefix. rUrrNz FSDP assumes z2 is in the state_dict but the state_dict only has z . prefix=z, module_name=z param_name=z rank=_has_been_clonedFTz#Failed to clone() tensor with name z. This may mean that this state_dict entry could point to invalid memory regions after returning from state_dict() call if this parameter is managed by FSDP. Please check clone implementation of z . Error: r)!rFSDP_WRAPPED_MODULErrRrlrTr r>rrprcrrrS startswithrr?keysrrrlclonerirrbrrrrrrrre) rrrJrprfrrsrWZ clean_keyZ clean_prefixrcr]r]r^r!}sN       .      z3FullyShardedDataParallel._full_post_state_dict_hookcCst||td||jjs"|St|jtd}|dk s>> # xdoctest: +SKIP("undefined variables") >>> import torch >>> from torch.distributed.fsdp import FullyShardedDataParallel as FSDP >>> from torch.distributed.fsdp import StateDictType >>> torch.cuda.set_device(device_id) >>> my_module = nn.Linear(...) >>> sharded_module = FSDP(my_module) >>> full_state_dict_config = FullStateDictConfig(offload_to_cpu=True, rank0_only=True) >>> with FSDP.state_dict_type(sharded_module, StateDictType.FULL_STATE_DICT, full_state_dict_config): >>> full_dict = sharded_module.state_dict() >>> full_dict.keys() >>> odict_keys(['weight', 'bias']) >>> # using local state dict >>> with FSDP.state_dict_type(sharded_module, StateDictType.LOCAL_STATE_DICT): >>> local_dict = sharded_module.state_dict() >>> local_dict.keys() >>> odict_keys(['flat_param', 'inner.flat_param']) .. warning:: This needs to be called on all ranks, since synchronization primitives may be used. NF)rMrrprq)rdrMrz_sharded_state_dict/local_state_dict can only be called when parameters are flatten and sharded.zUnknown StateDictType rU)!rcrrrwrrrLrmrrNrqrprrRrlr contextlibsuppressrrrrrrrrnrorTrrrrr=)rrRrZfull_state_dict_configrqrpZ summon_ctxrr3r]r^r' sX)       z#FullyShardedDataParallel.state_dict)rRrr}c Os2||tj|j||W5QRSQRXdS)z Returns the local state of the module. Parameters are flattened and sharded, so the resulting state_dict can only be loaded after the module has been wrapped with FSDP. N)rrLrnrrrRrr]r]r^_local_state_dict sz*FullyShardedDataParallel._local_state_dictcOs<|tjgt|dddk s"t|jdddd|_dS)N_full_param_ctx)rrRrlrlr?r__exit__rr]r]r^r+ sz8FullyShardedDataParallel._full_post_load_state_dict_hookc Os6|tj |j|f||W5QRSQRXdS)a Returns the sharded states of the module. Parameters are unflattened and sharded, so the resulting state_dict can be used with any parallelism (e.g., DPP, model parallelism, and single trainer) after a valid resharding. N)set_state_dict_typerLrorrr]r]r^_sharded_state_dict sz,FullyShardedDataParallel._sharded_state_dictcCsHt|dddkst|jddd|_|jt|||tddS)NrFTrrU)rlr?rr __enter__rr)rrrJr]r]r^r& s z7FullyShardedDataParallel._full_pre_load_state_dict_hookcOsdSrr]rr]r]r^r, sz9FullyShardedDataParallel._local_post_load_state_dict_hookcCst|||td|tdt}||krNt|jtddksJtddS||}t|tshtd|}t |stdt t j |dj }|jj}|dk st|jd|fkr||kstd|d|dt|d|jg}|||<dS) z This hook finds the local flat_param for this FSDP module from the state_dict. The flat_param should be a ShardedTensor. This hook converts the ShardedTensor to a tensor. No copy happen unless padding is required. rUNzTNo flat parameter in state_dict but self._fsdp_wrapped_module.flat_param is not Nonez4Tensors in local_state_dict should be ShardedTensor.z9load_local_state_dict assume one shard per ShardedTensor.rzLocal shard size = z% and the tensor in the state_dict is )rrr;rlrTr?rrrrrrcrrrrrFpad)rrrJrZ load_tensorshardsrr]r]r^r' s2  z8FullyShardedDataParallel._local_pre_load_state_dict_hookcOsdSrr]rr]r]r^r- sz;FullyShardedDataParallel._sharded_post_load_state_dict_hookcCs2t|||td|jjs"dS|jjjs4tdg}|jjjjD]L\}}}| |}|td||}| |}t |\}} t | dkst dt | d|j|} |d} t| |j| | } | r0ttj| dj} | js| } | | }|dkrDt| d|g} ntj| |jd} tj| |j| jd}t j!|| |j"d |#dd| $|}|%|qD|jj}t&j'|d d }t&(||j|j\}}|)|j*||kst d |d |d|j+|ks t d|d |j+d|||d<dS)z The hook combines the unflattened, sharded parameters (ShardedTensor) to a new FlatParameter and shards the new FlatParameter to the local chunk. rUNzSload_sharded_state_dict can only be called when parameters are flatten and sharded.rz&Expects 0 or 1 shard per rank but got z shards on rank rrrF)rz+The loaded local chunk has different numel(z) from the local chunk z-The loaded local chunk has different padding(z_fsdp_wrapped_module.flat_param),rrrTr rrrrZ _param_infosrr~r.rr?rrrmathceilrrrcrrflattenZis_cudarrrrrdemptyrrrrZreshaperr7Zflatten_paramsZ _get_shardrerr)rrrJZnonsharded_tensorsrsrrWrrHrZ param_numelZ dim_0_size chunk_sizeZ local_tensorZ num_paddingrrZloaded_flat_paramZ num_to_padr]r]r^r( sd            z:FullyShardedDataParallel._sharded_pre_load_state_dict_hookcGs4tt|}tjrtj|j|j||dS)z ``_pre_state_dict_hook` is called before ``self._load_from_state_dict()`` is called. ``self._state_dict_type`` is used to decide what preprocessing will be done. N)rrGrcrrrwr)r)rrrJrRrr]r]r^r%( s   z2FullyShardedDataParallel._pre_load_state_dict_hook)rrRr}cGstt|}|j|jdSr)rrGr.r)rrRrr]r]r^r*; s z3FullyShardedDataParallel._post_load_state_dict_hook)rr}cstj|f|S)a The entry point of all three FSDP ``load_state_dict`` APIs. By default, calling ``load_state_dict`` on an FSDP module will result in FSDP attempting to load a "full" state_dict, i.e. a state_dict consisting of full, unsharded, unflattened original module parameters. This requires FSDP to load the full parameter context on each rank which could result in GPU OOM. As a result, :func:`state_dict_type` API is available to configure between ``load_state_dict`` implementations. User can thus use ``with self.state_dict_type(self, StateDictType.LOCAL_STATE_DICT)`` context manager to load a local state dict checkpoint that will restore only local shards of the module. Currently, the only supported implementations are ``StateDictType.LOCAL_STATE_DICT`` and ``StateDictType.FULL_STATE_DICT`` (default). Please see :func:`state_dict` for documentation around creating an FSDP checkpoint. Example:: >>> # xdoctest: +SKIP("undefined variables") >>> import torch >>> from torch.distributed.fsdp import FullyShardedDataParallel as FSDP >>> from torch.distributed.fsdp import StateDictType >>> torch.cuda.set_device(device_id) >>> my_module = nn.Linear(...) >>> sharded_module = FSDP(my_module) >>> checkpoint = torch.load(PATH) >>> full_state_dict = checkpoint['full_state_dict'] >>> with FSDP.state_dict_type(sharded_module, StateDictType.FULL_STATE_DICT): >>> sharded_module.load_state_dict(full_state_dict) >>> full_dict.keys() >>> odict_keys(['weight', 'bias']) >>> # using local state dict >>> local_state_dict = checkpoint['local_state_dict'] >>> with FSDP.state_dict_type(sharded_module, StateDictType.LOCAL_STATE_DICT): >>> sharded_module.load_state_dict(local_state_dict) >>> local_dict.keys() >>> odict_keys(['flat_param', 'inner.flat_param']) .. warning:: This needs to be called on all ranks, since synchronization primitives may be used. )rload_state_dict)rrrRrr3r]r^rC s.z(FullyShardedDataParallel.load_state_dictc Gs6||tj|j|f|W5QRSQRXdS)zI Load states from a flattened, sharded state dictionary. N)rrLrnr)rrrRr]r]r^_load_local_state_dicts sz/FullyShardedDataParallel._load_local_state_dictzOrderedDict[str, torch.Tensor])rstrictr}c Cs0|tj|||W5QRSQRXdS)zK Load states from a unflattened, sharded state dictionary. N)rrLror)rrrr]r]r^_load_sharded_state_dict~ sz1FullyShardedDataParallel._load_sharded_state_dictc stjjdj||\}}d}tjjj d}fddj D}tj j |} j |||j D]*}t |j jjkdjd|j jq|j||}j ||||W5QRSQRXdS)z Runs the forward pass for the wrapped module, inserting FSDP-specific pre- and post-forward sharding logic. FullyShardedDataParallel.forwardNrcs"g|]}j o|jjtjkqSr])r_configrr9rZrrr]r^r sz4FullyShardedDataParallel.forward..z5Expected `FlatParameter` to be on the compute device r)rcautogradprofilerrecord_functionr_fsdp_root_pre_forwardrUrV_pre_forward_unshardrr _pre_forwardr5rrrrT _post_forward) rrRrunused unshard_fnr{ reshard_fnroutputr]rr^forward s*    r)rrrinputcCsFtj|_|j||j|D] }tj|_q|dk r8|||dS)a Runs the pre-forward logic. This includes an opportunity to unshard currently sharded parameters such as those for the current forward and registering post-backward hooks for these current parameters. Args: handles (List[FlatParamHandle]): Handles giving the parameters used in the current forward. unshard_fn (Optional[Callable]): A callable to unshard any currently sharded parameters or ``None`` to not do any unsharding. module (nn.Module): Unused; expected by the hook signature. input (Any): Unused; expected by the hook signature. N) rRrkrrrZtrainingr:r_register_post_backward_hooks)rrrrrrr]r]r^r s z%FullyShardedDataParallel._pre_forwardcCsD|r@||t|}d|j|<tj|jd||dS)z'Unshards parameters in the pre-forward.FrvN) rzrrrcrrryrr)rrrr]r]r^r s   z-FullyShardedDataParallel._pre_forward_unshard)rrrrrr}cCsD|j||dk r||||}tj|_|D] }tj|_q2|S)a Runs the post-forward logic. This includes an opportunity to reshard currently unsharded parameters such as those used in the current forward and registering pre-backward hooks on the forward outputs. Args: handles (List[FlatParamHandle]): Handles giving the parameters used in the current forward. reshard_fn (Optional[Callable]): A callable to reshard any currently unsharded parameters (e.g. from the current forward) or ``None`` to not do any resharding. module (nn.Module): Unused; expected by the hook signature. input (Any): Unused; exepcted by the hook signature. output (Any): Forward pass output; pre-backward hooks are registered on the tensors that require gradients in this output. Postcondition: Each ``FlatParameter`` 's data points to the sharded flattened parameter. N)rr_register_pre_backward_hooksrRrjrr:r)rrrrrrrr]r]r^r s   z&FullyShardedDataParallel._post_forwardcOsTt|||jjd\}}|d}|d}|rL|jj}|j|f||\}}||fS)ziMoves the forward inputs to the compute device and casts them to the appropriate dtype if needed.Fr)r rrrrr_r)rrRrZ input_dtyper]r]r^_cast_forward_inputs sz-FullyShardedDataParallel._cast_forward_inputscOslt|jdk d|js||fS|jrL||D]}t|j}|r.d|j|<q.||j||\}}||fS)am Runs pre-forward logic specific to the root FSDP instance, which should run before any individual module's pre-forward. This includes synchronizing with the previous iteration and casting the forward inputs appropriately. If this is called on a non-root FSDP instance, then the forward inputs are returned directly. Nz$Expects a root FSDP to have been setT) r5rrrrrrrr)rrRrrrr]r]r^r s  z/FullyShardedDataParallel._fsdp_root_pre_forward)rMrrqrpr}c csNtj|dd}t.}|D]}||j||||dqdVW5QRXdS)a A context manager to expose full params for FSDP instances. Can be useful *after* forward/backward for a model to get the params for additional processing or checking. It can take a non-FSDP module and will summon full params for all contained FSDP modules as well as their children, depending on the ``recurse`` argument. .. note:: This can be used on inner FSDPs. .. note:: This can *not* be used within a forward or backward pass. Nor can forward and backward be started from within this context. .. note:: Parameters will revert to their local shards after the context manager exits, storage behavior is the same as forward. .. note:: The full parameters can be modified, but only the portion corresponding to the local param shard will persist after the context manager exits (unless ``writeback=False``, in which case changes will be discarded). In the case where FSDP does not shard the parameters, currently only when ``world_size == 1``, or ``NO_SHARD`` config, the modification is persisted regardless of ``writeback``. .. note:: This method works on modules which are not FSDP themselves but may contain multiple independent FSDP units. In that case, the given arguments will apply to all contained FSDP units. .. warning:: Note that ``rank0_only=True`` in conjunction with ``writeback=True`` is not currently supported and will raise an error. This is because model parameter shapes would be different across ranks within the context, and writing to them can lead to inconsistency across ranks when the context is exited. .. warning:: Note that ``offload_to_cpu`` and ``rank0_only=False`` will result in full parameters being redundantly copied to CPU memory for GPUs that reside on the same machine, which may incur the risk of CPU OOM. It is recommended to use ``offload_to_cpu`` with ``rank0_only=True``. Args: recurse (bool, Optional): recursively summon all params for nested FSDP instances (default: True). writeback (bool, Optional): if ``False``, modifications to params are discarded after the context manager exits; disabling this can be slightly more efficient (default: True) rank0_only (bool, Optional): if ``True``, full parameters are materialized on only global rank 0. This means that within the context, only rank 0 will have full parameters and the other ranks will have sharded parameters. Note that setting ``rank0_only=True`` with ``writeback=True`` is not supported, as model parameter shapes will be different across ranks within the context, and writing to them can lead to inconsistency across ranks when the context is exited. offload_to_cpu (bool, Optional): If ``True``, full parameters are offloaded to CPU. Note that this offloading currently only occurs if the parameter is sharded (which is only not the case for world_size = 1 or ``NO_SHARD`` config). It is recommended to use ``offload_to_cpu`` with ``rank0_only=True`` to avoid redundant copies of model parameters being offloaded to the same CPU memory. TrrMrrqrpN)rGrr ExitStack enter_contextr)rrMrrqrpZroot_fsdp_modulesstackr]r]r^summon_full_params" s A  z+FullyShardedDataParallel.summon_full_paramsrc cs|r|rtd|r"|s"td|rlt4}||D]}||jd|||dq:dVW5QRXdStj | | t jg|jD]}|jtjkstqt j|_|jD] }tj|_qdd|jD}||jtj |jd|rD|jdkrD||j|z dVW5t j|_|jD]}tj|_q0Xnt}|jD]"}|rT|jrT||qT||jz dVW5||r| |j||j|t j|_|jD]}tj|_qXW5QRXdS) Nzwriteback=True and rank0_only=True is not supported, as model parameter shapes will be different across ranks, and writing to them can lead to inconsistencies across ranks when the context is exited.zoffload_to_cpu and rank0_only=False will result in full parameters being redundantly copied to CPU memory for GPUs that reside on the same machine, which may incur the risk of CPU OOM. It is recommended to use ``offload_to_cpu`` with rank0_only=True.FrcSsg|] }|qSr])Z needs_unshardrr]r]r^r sz@FullyShardedDataParallel._summon_full_params..rvr)!r=rrrrrrrrcrrwrrrRrjrrr:r?rlrrzrryrrrrZto_cpurTZunflatten_as_paramsclose_write_back_to_local_shard) rrMrrqrprrrr{r]r]r^rv sj             z,FullyShardedDataParallel._summon_full_paramsrcCsf|D]\}|jsq|jjdks.td|jjt|j|j|j\}}|jj d|  |qdS)a For each handle, writes back the this rank's shard of the unsharded flattened parameter to the sharded flattened parameter. Precondition: Each handle's ``FlatParameter`` 's data points to the padded unsharded flattened parameter. r"z-Expects `flat_param` to be flattened but got N) rrndimr?shaper7Z_get_unpadded_shardrrrrcopy_)rrrr rr]r]r^r s   z3FullyShardedDataParallel._write_back_to_local_shardc/sB|jtjk}tj||D]"\}}|r2|td}||fVqdS)z Overrides :meth:`named_buffers()` to intercept buffer names and remove all occurrences of the FSDP-specific flattened buffer prefix when inside the :meth:`summon_full_params` context manager. rN)rrRrlrrNr FSDP_PREFIX)rrRrin_summon_full_paramsrPror3r]r^rN s  z&FullyShardedDataParallel.named_buffersc/sB|jtjk}tj||D]"\}}|r2|td}||fVqdS)z Overrides :meth:`named_parameters()` to intercept parameter names and remove all occurrences of the FSDP-specific flattened parameter prefix when inside the :meth:`summon_full_params` context manager. rN)rrRrlrrrrr)rrRrrrsrHr3r]r^rr s  z)FullyShardedDataParallel.named_parameters)outputsrr}cszts |Sjrd_tr8dj<dj<ttt ddfdd tj tj dfdd }t ||S) aT Registers pre-backward hooks on the tensors that require gradients in the forward pass outputs ``outputs``, which were computed using the ``FlatParameter`` s of ``handles``. Returns: Forward pass outputs with pre-backward hooks registered to tensors that require gradients. FN)rrr}c st|}|rj|drdStjjdjrDjsD n|rV t j gt j _|spW5QRdS|D] }tj |_qt|tjjddj|<||D] }|qdj|<W5QRXdS)zRPrepares ``_handles`` 's ``FlatParameter`` s for gradient computation.FNz+FullyShardedDataParallel._pre_backward_hookrvT)rrrrcrrrr_post_backward_callback_queued_queue_wait_for_post_backwardrrRrjrhrr:rrzrrryrrrZprepare_gradient)rrZ _handles_keyrrr]r^_pre_backward_hook s,       zQFullyShardedDataParallel._register_pre_backward_hooks.._pre_backward_hook)tr}cs&|jr"|tdj<|S)NT)r register_hookrUrVr)rrrrrr]r^_register_hookH s zMFullyShardedDataParallel._register_pre_backward_hooks.._register_hook) rcis_grad_enabledrrrrrr r7rrr0)rrrr r]r r^r s  )z5FullyShardedDataParallel._register_pre_backward_hookscCs~ts dS|D]h}|j}t|d}|s|js0q||}t|jdk d|jjdd}| t |j |}||f|_ qdS)aC Registers post-backward hooks on the ``FlatParameter`` s' ``AccumulateGrad`` objects to reshard and to reduce-scatter gradients. The ``AccumulateGrad`` object represents the last function that finalizes the ``FlatParameter`` 's gradient, so it only runs after its entire gradient computation has finished. We register the post-backward hook only once in the *first* forward that a ``FlatParameter`` participates in. This relies on the ``AccumulateGrad`` object being preserved through multiple forwards. N_post_backward_hook_statezZThe `grad_fn` is needed to access the `AccumulateGrad` and register the post-backward hookr)rcr rr>rZ expand_asr5Zgrad_fnZnext_functionsr rUrV_post_backward_hookr )rrrrZalready_registeredZtemp_flat_paramZacc_gradZ hook_handler]r]r^rP s"    z6FullyShardedDataParallel._register_post_backward_hooks)rrr}c Gs|j}d|_tjjd|tjtj gtj |_ t j |_ | rZ|jrZ|j||jdkrrW5QRdS|jjrtd||}||g|g|f}|||jsW5QRdS|jdtjtj|jd|jj}|r|s|jj |j!j"|j_|j#j$rHt%|j&dk dt%|j'dk d|jj}|j(rFd|_t)|}t*|+|j,} |j,| d-|-} t./|d| g} t0| d} |&|j'| | |1| |t2|d } | r8t%|j3j4| j4kd |j3j4d | j4t%|j3j5| j5kd |j3j5d | j5|j3| 7_3n| |_3|j3}n6|j6t7j8krd|&|j'|j|9s||1|j||j:j;r|jdd|j?tj|?|jdW5QRXW5QRXdS)a Reduce-scatters the gradient of ``handle`` 's ``FlatParameter``. Precondition: The ``FlatParameter`` 's ``.grad`` attribute contains the unsharded gradient for the local batch. Postcondition: - If using ``NO_SHARD``, then the ``.grad`` attribute is the reduced unsharded gradient. - Otherwise, the ``_saved_grad_shard`` attribute is the reduced sharded gradient (accumulating with any existing gradient). T,FullyShardedDataParallel._post_backward_hookNz;FSDP only works with gradients that don't require gradientsrz%Communication hook should not be Nonez+Communication hook state should not be Noner_saved_grad_shardz@Shape mismatch when accumulating gradients: existing grad shape=z new grad shape=zBDevice mismatch when accumulating gradients: existing grad device=z new grad device=)Z non_blocking)@rrrcrrrrrRrhrirr:r_use_param_exec_order_policy_param_exec_order_prep_stage_fsdp_params_exec_orderrgradrr!_should_free_unsharded_flat_paramrrrrryrrrxrrrrerr`rrr5rrrrrchunkrrrrr_cast_grad_to_param_dtyper>rrrrrHr\rrrgrrri record_stream)rrrrHrrZorig_grad_datarZ grad_flattenchunksZnum_padZinput_flattenedrZaccumulate_gradr]r]r^rv s              r)rrHcCsP|tj|sL|s$|rL|j}|jj|jd|_| t j dS)a! Casts gradient ``grad`` back to the full parameter dtype so that the optimizer step runs with that dtype. This performs an actual cast if 1. parameters were in reduced precision during the forward since then gradients would be in that reduced precision, or 2. parameters were not in reduced precision but gradients were in reduced precision for communication. However, if a low precision communication hook is registered, then this dtype cast happens in the hook instead. rN) rrRrirrrrrerdrrcrr)rrrHZlow_prec_grad_datar]r]r^r s z2FullyShardedDataParallel._cast_grad_to_param_dtype)rcCs|jr |jp|jjtjkSr)rrrrr9rZrtr]r]r^r: s  z:FullyShardedDataParallel._should_free_unsharded_flat_paramcCs<t|jd|jrdS|tjgd|_tj|j dS)z Queues a post-backward callback from the root FSDP instance, which should happen at the beginning of its pre-backward. zL`_queue_wait_for_post_backward()` should be called on the root FSDP instanceNT) r5rrrrRrjrZ_execution_engineZqueue_callback_wait_for_post_backwardrr]r]r^r@ sz6FullyShardedDataParallel._queue_wait_for_post_backwardcsjstdjr@tjjdjj r@tj j t ddfdd }t ddfdd }D]P}|||||jtj|_|jD] }tj|_q|j|jr|d _q|rjrdS) z?Wait for post-backward to finish. Only called on root instance.z3_wait_for_post_backward can only be called on root.rN)rr}c szZg}g}|jD]:}|j|jjk}|r0q||||q||WnDtk r}z&tdd|dt |dd|W5d}~XYnXdS)a Reshards full parameters that may have not been resharded in post_backward_hook. This can happen when an FSDP module's output is used in forward so its pre-backward fires unsharding the param, but post-backward does not fire since the output was not ultimately used in loss computation so FSDP parameter did not get a gradient. Fz&Got exception while resharding module rg)Zraise_assertion_errorN) rrZdata_ptrrrrr Exceptionr5r)rr{Zhandles_to_reshardrZalready_reshardedrcrr]r^_catch_all_reshardd s$   zLFullyShardedDataParallel._wait_for_post_backward.._catch_all_reshardcs4|jD]&}|j}|jrt|drNtt|jdkd|jdt|dj sVqt|drt|j t dkd|j d|j |j |_ nrt|d rt|j |jj kd|j d |jj |jr|j|_ |r|j |jj|j _nt|j p|j d t|d rt|d tt|d d d|_qdS)z&Helper used below on all fsdp modules.r rz1p._post_backward_hook_state fields are not valid.r"rrzDevice mismatch: p=z p._cpu_grad=rz p._saved_grad_shard=zNAll sharded parameters that received a gradient should use `_saved_grad_shard`rz7Expected flag _post_backward_called to be set on param.FN)rrrr>r5rr removedelattrrrrcrrrrrrerr_rr)rrrDrr]r^_finalize_params sP           zJFullyShardedDataParallel._wait_for_post_backward.._finalize_paramsF)rr?rrcrrryrrrgrwrrrGrrrrRrjrrr:rrrrr)_param_exec_order_policy_second_iter_init)rrrrCrr]rr^rO s( "<    z0FullyShardedDataParallel._wait_for_post_backwardcCs^d|_|j|D]@}||k rt|trt|ds@tdt|dsRtdd|_qdS)NF_param_exec_order_policyzINon-root FSDP modules should also have _param_exec_order_policy attributerzMNon-root FSDP modules should also have _param_exec_order_prep_stage attribute)rrreverser6rrGr>r?)rrCr]r]r^r  s"  zBFullyShardedDataParallel._param_exec_order_policy_second_iter_init)stater}cCsft|tr|g}|j|krbd|d|j}|jdkrZtd|td|tt|dS)z!Assert we are in the given state.zexpected to be in states z but current state is rzAsserting FSDP instance is: zERROR: N)rrRrrprint traceback print_stackr=)rr#msgr]r]r^r s   z&FullyShardedDataParallel._assert_statec cs||jstd|tjg}|D]$}t|tr.| ||j fd|_ q.z dVW5|D]\}}|j rztd||_ qdXdS)a8 A context manager to disable gradient synchronizations across FSDP instances. Within this context, gradients will be accumulated in module variables, which will later be synchronized in the first forward-backward pass after exiting the context. This should only be used on the root FSDP instance and will recursively apply to all children FSDP instances. .. note:: This likely results in higher memory usage because FSDP will accumulate the full model gradients (instead of gradient shards) until the eventual sync. .. note:: When used with CPU offloading, the gradients will not be offloaded to CPU when inside the context manager. Instead, they will only be offloaded right after the eventual sync. z4`no_sync()` on inner FSDP instances is not supportedFzX`_sync_gradients` was incorrectly set to `True` while in the `no_sync()` context managerN) rrr?rrRrjr6rrGrr)rZ old_flagsrCZold_flagr]r]r^no_syncs     z FullyShardedDataParallel.no_synccCsdd|DS)z[ Recursively returns a list of all module parameters that have a gradient. cSsg|]}|jdk r|qSrrr\r]r]r^r*s z=FullyShardedDataParallel.params_with_grad..)rBrr]r]r^params_with_grad%sz)FullyShardedDataParallel.params_with_grad@)max_norm norm_typer}cCs|||jstd|tjt|}t|}t|j | }|t j krr|}t j|tjjj|jdn$||}t j||jd|d|}|jr|}tj||j|jd|d}|dkr|j D],}|jdk st|j||jjqdS) a Clip all gradients at this point in time. The norm is computed over all gradients together, as if they were concatenated into a single vector. Gradients are modified in-place. Args: max_norm (float or int): max norm of the gradients norm_type (float or int): type of the used p-norm. Can be ``'inf'`` for infinity norm. Returns: Total norm of the parameters (viewed as a single vector). .. note:: This is analogous to ``torch.nn.utils.clip_grad_norm_`` but handles the partitioning and multiple devices per rank under the hood. The default torch util is not applicable here, because each rank only has a partial view of all the grads in the model, so calling it for FSDP models would lead to different scaling being applied per subset of model parameters. .. warning:: This needs to be called on all ranks, since synchronization primitives will be used. zBclip_grad_norm should only be called on the root (parent) instance)oprrg?rgư>r"N)rrrr?rrRrjfloat_calc_grad_normr*rrinfrZ all_reducerc distributedZReduceOpMAXrrrrrdrrriZmul_re)rr,r- local_normZ total_normZ clip_coefrDr]r]r^clip_grad_norm_,s(    z(FullyShardedDataParallel.clip_grad_norm_cCs|dk rtddS)NzThe `optim_input` argument is deprecated and will be removed after PyTorch 1.13. You may remove it from your code without changing its functionality.)rr) optim_inputr]r]r^_warn_optim_inputbsz*FullyShardedDataParallel._warn_optim_inputcCs$|dkr|dkrdS|dk r dSdS)NTFr])r6optimr]r]r^_is_using_optim_inputjs z.FullyShardedDataParallel._is_using_optim_input)modelr8r6rqrr}c Cs,t|t||}t||||d||dS)aD Consolidates the full optimizer state on rank 0 and returns it as a :class:`dict` following the convention of :meth:`torch.optim.Optimizer.state_dict`, i.e. with keys ``"state"`` and ``"param_groups"``. The flattened parameters in ``FSDP`` modules contained in ``model`` are mapped back to their unflattened parameters. .. warning:: This needs to be called on all ranks since synchronization primitives are used. However, if ``rank0_only=True``, then the state dict is only populated on rank 0, and all other ranks return an empty :class:`dict`. .. warning:: Unlike ``torch.optim.Optimizer.state_dict()``, this method uses full parameter names as keys instead of parameter IDs. .. note:: Like in :meth:`torch.optim.Optimizer.state_dict`, the tensors contained in the optimizer state dict are not cloned, so there may be aliasing surprises. For best practices, consider saving the returned optimizer state dict immediately, e.g. using ``torch.save()``. Args: model (torch.nn.Module): Root module (which may or may not be a :class:`FullyShardedDataParallel` instance) whose parameters were passed into the optimizer ``optim``. optim (torch.optim.Optimizer): Optimizer for ``model`` 's parameters. optim_input (Optional[Union[List[Dict[str, Any]], Iterable[torch.nn.Parameter]]]): Input passed into the optimizer ``optim`` representing either a :class:`list` of parameter groups or an iterable of parameters; if ``None``, then this method assumes the input was ``model.parameters()``. This argument is deprecated, and there is no need to pass it in anymore. (Default: ``None``) rank0_only (bool): If ``True``, saves the populated :class:`dict` only on rank 0; if ``False``, saves it on all ranks. (Default: ``True``) group (dist.ProcessGroup): Model's process group or ``None`` if using the default process group. (Default: ``None``) Returns: Dict[str, Any]: A :class:`dict` containing the optimizer state for ``model`` 's original unflattened parameters and including keys "state" and "param_groups" following the convention of :meth:`torch.optim.Optimizer.state_dict`. If ``rank0_only=True``, then nonzero ranks return an empty :class:`dict`. Fr:r8r6rq shard_staterusing_optim_inputrGr7r9r*)r:r8r6rqrr=r]r]r^full_optim_state_dictus8 z.FullyShardedDataParallel.full_optim_state_dict)r:r8r6rr}c Cs,t|t||}t|||dd||dS)a+ The API is similar to :meth:`full_optim_state_dict` but this API chunks all non-zero-dimension states to :class:`ShardedTensor` to save memory. This API should only be used when the model ``state_dict`` is derived with the context manager ``with state_dict_type(SHARDED_STATE_DICT):``. For the detailed usage, refer to :meth:`full_optim_state_dict`. .. warning:: The returned state dict contains ``ShardedTensor`` and cannot be directly used by the regular ``optim.load_state_dict``. FTr;r>)r:r8r6rr=r]r]r^sharded_optim_state_dicts  z1FullyShardedDataParallel.sharded_optim_state_dict)r?r:r6r8r}cCs2t|t||}t||d}t|||||S)a Shards the full optimizer state dict ``full_optim_state_dict`` by remapping the state to flattened parameters instead of unflattened parameters and restricting to only this rank's part of the optimizer state. The first argument should be the return value of :meth:`full_optim_state_dict`. Example:: >>> # xdoctest: +SKIP("undefined variables") >>> from torch.distributed.fsdp import FullyShardedDataParallel as FSDP >>> model, optim = ... >>> full_osd = FSDP.full_optim_state_dict(model, optim) >>> torch.save(full_osd, PATH) >>> # Define new model with possibly different world size >>> new_model, new_optim = ... >>> full_osd = torch.load(PATH) >>> sharded_osd = FSDP.shard_full_optim_state_dict(full_osd, new_model) >>> new_optim.load_state_dict(sharded_osd) .. note:: Both :meth:`shard_full_optim_state_dict` and :meth:`scatter_full_optim_state_dict` may be used to get the sharded optimizer state dict to load. Assuming that the full optimizer state dict resides in CPU memory, the former requires each rank to have the full dict in CPU memory, where each rank individually shards the dict without any communication, while the latter requires only rank 0 to have the full dict in CPU memory, where rank 0 moves each shard to GPU memory (for NCCL) and communicates it to ranks appropriately. Hence, the former has higher aggregate CPU memory cost, while the latter has higher communication cost. Args: full_optim_state_dict (Dict[str, Any]): Optimizer state dict corresponding to the unflattened parameters and holding the full non-sharded optimizer state. model (torch.nn.Module): Root module (which may or may not be a :class:`FullyShardedDataParallel` instance) whose parameters correspond to the optimizer state in ``full_optim_state_dict``. optim_input (Optional[Union[List[Dict[str, Any]], Iterable[torch.nn.Parameter]]]): Input passed into the optimizer representing either a :class:`list` of parameter groups or an iterable of parameters; if ``None``, then this method assumes the input was ``model.parameters()``. This argument is deprecated, and there is no need to pass it in anymore. (Default: ``None``) optim (Optional[torch.optim.Optimizer]): Optimizer that will load the state dict returned by this method. This is the preferred argument to use over ``optim_input``. (Default: ``None``) Returns: Dict[str, Any]: The full optimizer state dict now remapped to flattened parameters instead of unflattened parameters and restricted to only include this rank's part of the optimizer state. TrGr7r9r%r,)r?r:r6r8r= sharded_osdr]r]r^shard_full_optim_state_dicts"A z4FullyShardedDataParallel.shard_full_optim_state_dict)r@r:r6r8r}cCs4t|t||}t||dd}t|||||S)a The API is similar to :meth:`shard_full_optim_state_dict`. The only difference is that the input ``sharded_optim_state_dict`` should be returned from :meth:`sharded_optim_state_dict`. Therefore, there will be all-gather calls on each rank to gather ``ShardedTensor`` s. Args: sharded_optim_state_dict (Dict[str, Any]): Optimizer state dict corresponding to the unflattened parameters and holding the sharded optimizer state. model (torch.nn.Module): Refer to :meth:``shard_full_optim_state_dict``. Returns: Refer to :meth:`shard_full_optim_state_dict`. Tr:r<rA)r@r:r6r8r=Z flattened_osdr]r]r^ flatten_sharded_optim_state_dict3s" z9FullyShardedDataParallel.flatten_sharded_optim_state_dict)r?r:r6r8rr}c Cst|t||}|dkr.t|dr.|j}t|}t|}tj |}t j rbt dnt d} |rt j std|dkr|dkrtdt||dd } t| |} t|dkr| nd||} t| |dkr| nd|||| } t| ||||} | S) a Scatters the full optimizer state dict from rank 0 to all other ranks, returning the sharded optimizer state dict on each rank. The return value is the same as :meth:`shard_full_optim_state_dict`, and on rank 0, the first argument should be the return value of :meth:`full_optim_state_dict`. Example:: >>> # xdoctest: +SKIP("undefined variables") >>> from torch.distributed.fsdp import FullyShardedDataParallel as FSDP >>> model, optim = ... >>> full_osd = FSDP.full_optim_state_dict(model, optim) # only non-empty on rank 0 >>> # Define new model with possibly different world size >>> new_model, new_optim, new_group = ... >>> sharded_osd = FSDP.scatter_full_optim_state_dict(full_osd, new_model, group=new_group) >>> new_optim.load_state_dict(sharded_osd) .. note:: Both :meth:`shard_full_optim_state_dict` and :meth:`scatter_full_optim_state_dict` may be used to get the sharded optimizer state dict to load. Assuming that the full optimizer state dict resides in CPU memory, the former requires each rank to have the full dict in CPU memory, where each rank individually shards the dict without any communication, while the latter requires only rank 0 to have the full dict in CPU memory, where rank 0 moves each shard to GPU memory (for NCCL) and communicates it to ranks appropriately. Hence, the former has higher aggregate CPU memory cost, while the latter has higher communication cost. Args: full_optim_state_dict (Optional[Dict[str, Any]]): Optimizer state dict corresponding to the unflattened parameters and holding the full non-sharded optimizer state if on rank 0; the argument is ignored on nonzero ranks. model (torch.nn.Module): Root module (which may or may not be a :class:`FullyShardedDataParallel` instance) whose parameters correspond to the optimizer state in ``full_optim_state_dict``. optim_input (Optional[Union[List[Dict[str, Any]], Iterable[torch.nn.Parameter]]]): Input passed into the optimizer representing either a :class:`list` of parameter groups or an iterable of parameters; if ``None``, then this method assumes the input was ``model.parameters()``. This argument is deprecated, and there is no need to pass it in anymore. (Default: ``None``) optim (Optional[torch.optim.Optimizer]): Optimizer that will load the state dict returned by this method. This is the preferred argument to use over ``optim_input``. (Default: ``None``) group (dist.ProcessGroup): Model's process group or ``None`` if using the default process group. (Default: ``None``) Returns: Dict[str, Any]: The full optimizer state dict now remapped to flattened parameters instead of unflattened parameters and restricted to only include this rank's part of the optimizer state. Nrrrz#NCCL requires a GPU for collectivesrz1Rank 0 must pass in the full optimizer state dictFrD)rGr7r9r>rrZget_rankZget_world_sizeZdistributed_c10dZ_check_for_nccl_backendrcrrrrr=r%r+r$r#r,) r?r:r6r8rr=rrZ using_ncclZbroadcast_deviceZflat_osdZ processed_osdrBr]r]r^scatter_full_optim_state_dict^sZA     z6FullyShardedDataParallel.scatter_full_optim_state_dict)optim_state_dictoptim_state_key_typer:r6r8r}cst|t||}|tjtjfks*t|}dd|dD}dd|dD}t|rbt|rrt|rt|sd|d } t | |tjkrt|s|tjkrt|r|Si} |tjkrV|rt ||nt |} t |fdd| Dfdd|dD| d<t|d | d <| d D]$} tfd d| d D| d <q,| S|tjkrt|} |rzt||nt|fd d| Dfd d|dD| d<t|d | d <| d D]$} tfdd| d D| d <q| S| S)aO Re-keys the optimizer state dict ``optim_state_dict`` to use the key type ``optim_state_key_type``. This can be used to achieve compatibility between optimizer state dicts from models with FSDP instances and ones without. To re-key an FSDP full optimizer state dict (i.e. from :meth:`full_optim_state_dict`) to use parameter IDs and be loadable to a non-wrapped model:: >>> # xdoctest: +SKIP("undefined variables") >>> wrapped_model, wrapped_optim = ... >>> full_osd = FSDP.full_optim_state_dict(wrapped_model, wrapped_optim) >>> nonwrapped_model, nonwrapped_optim = ... >>> rekeyed_osd = FSDP.rekey_optim_state_dict(full_osd, OptimStateKeyType.PARAM_ID, nonwrapped_model) >>> nonwrapped_optim.load_state_dict(rekeyed_osd) To re-key a normal optimizer state dict from a non-wrapped model to be loadable to a wrapped model:: >>> # xdoctest: +SKIP("undefined variables") >>> nonwrapped_model, nonwrapped_optim = ... >>> osd = nonwrapped_optim.state_dict() >>> rekeyed_osd = FSDP.rekey_optim_state_dict(osd, OptimStateKeyType.PARAM_NAME, nonwrapped_model) >>> wrapped_model, wrapped_optim = ... >>> sharded_osd = FSDP.shard_full_optim_state_dict(rekeyed_osd, wrapped_model) >>> wrapped_optim.load_state_dict(sharded_osd) Returns: Dict[str, Any]: The optimizer state dict re-keyed using the parameter keys specified by ``optim_state_key_type``. cSsg|]}t|tkqSr])r:rrZ param_keyr]r]r^rszCFullyShardedDataParallel.rekey_optim_state_dict..r#cSsg|]}t|tkqSr])r:rrIr]r]r^rszInvalid parameter keys: csg|] }|qSr]r]rY)param_to_param_namer]r^rscsi|]\}}||qSr]r])rparam_id param_stateparam_id_to_param_namer]r^ szCFullyShardedDataParallel.rekey_optim_state_dict..Z param_groupscsg|] }|qSr]r])rrKrMr]r^r$sr cs"i|]\}}|kr||qSr]r])rrsrH)param_to_param_idr]r^rO2scsi|]\}}||qSr]r])rrsrLparam_name_to_param_idr]r^rO7scsg|] }|qSr]r])rrsrQr]r^r=s)rGr7r9rQrsrtr?r_allrr=r'r&_get_param_to_param_nameitemscopydeepcopysorted_get_param_name_to_paramr)r()rGrHr:r6r8r=ZosdZuses_param_name_maskZuses_param_id_mask error_msgZnew_osdZparam_id_to_paramZ param_groupZparam_name_to_paramr])rNrRrPrJr^rekey_optim_state_dicts*                z/FullyShardedDataParallel.rekey_optim_state_dictcCs|jtjkrtjStjSdS)zT Returns a default communication hook based on a sharding strategy. N)rrHr\rZreduce_scatter_hookZallreduce_hookrr]r]r^rDs z/FullyShardedDataParallel._get_default_comm_hookcCstj|jdS)zZ Returns a default communication hook state based on a sharding strategy. r)rZ DefaultStaterrr]r]r^rMsz5FullyShardedDataParallel._get_default_comm_hook_state)r#hookcCsf|std||D]F}|jr,tdd|_|j|ksTtd|jjd||_||_qdS)aV Registers a communication hook which is an enhancement that provides a flexible hook to users where they can specify how FSDP aggregates gradients across multiple workers. This hook can be used to implement several algorithms like `GossipGrad `_ and gradient compression which involve different communication strategies for parameter syncs while training with :class:`FullyShardedDataParallel`. .. warning :: FSDP communication hook should be registered before running an initial forward pass and only once. Args: state (object): Passed to the hook to maintain any state information during the training process. Examples include error feedback in gradient compression, peers to communicate with next in `GossipGrad `_, etc. It is locally stored by each worker and shared by all the gradient tensors on the worker. hook (Callable): Callable, which has one of the following signatures: 1) ``hook: Callable[torch.Tensor] -> None``: This function takes in a Python tensor, which represents the full, flattened, unsharded gradient with respect to all variables corresponding to the model this FSDP unit is wrapping (that are not wrapped by other FSDP sub-units). It then performs all necessary processing and returns ``None``; 2) ``hook: Callable[torch.Tensor, torch.Tensor] -> None``: This function takes in two Python tensors, the first one represents the full, flattened, unsharded gradient with respect to all variables corresponding to the model this FSDP unit is wrapping (that are not wrapped by other FSDP sub-units). The latter represents a pre-sized tensor to store a chunk of a sharded gradient after reduction. In both cases, callable performs all necessary processing and returns ``None``. Callables with signature 1 are expected to handle gradient communication for a `NO_SHARD` case. Callables with signature 2 are expected to handle gradient communication for sharded cases. z9register_comm_hook can only be called on a root instance.z.communication hook can be only registered onceTz0communication hook should be default, but it is z insteadN)rr?rrrrrVr)rr#r\rAr]r]r^register_comm_hookSs'z+FullyShardedDataParallel.register_comm_hookc Os|d}|d}t|dstts6|jdkstdnt|jtr|jj}t|}|D]}t|t rZtdqZt |||dNz| ||jj Wn2t k r}ztd|dW5d}~XYnXW5QRXn|jdkstd |j|d<|j||d |_d |_g|_trt|jtrt} ||D]} | | | j<q4|jD].}|| krL| |jD]} |j| qdqLd |_|D]4}||k rt|t r|j|_|j|_|j|_qdS) Nrrtracing_configz:tracing_config should be None when torch.fx is not enabledzAThe input module of _patch_tracer should not contain FSDP modules)tracerr5execution_infozNtracer.trace failed inside _init_param_exec_order_wrap_policy with the error: rUzGtracing_config should either be an instance of TracingConfig or be NoneTF)r>r?_TORCH_FX_AVAILr^rrDr_rEr6rGrFtraceZ concrete_argsrbrZ init_policyrr!rrdictrrZmodule_forward_orderr r) rrRrrrr_r`rCrcZmodule_to_fsdpwraprr]r]r^rst  "     z;FullyShardedDataParallel._init_param_exec_order_wrap_policycCst|do|jS)Nr!)r>r!rr]r]r^rs z5FullyShardedDataParallel._use_param_exec_order_policycCs8t|do|j}|s4|D]}t|drtdq|S)NrZ_params_exec_order_hook_handlez]When not in execution order prep stage, all _params_exec_order_hook_handle should be removed.)r>rrBr?)rZ is_prep_stagerDr]r]r^_is_param_exec_order_prep_stages  z8FullyShardedDataParallel._is_param_exec_order_prep_stage) NNNNNNNNNFFF)F)NNNT)N)T)TTFF)TTFF)r+)NTN)NN)NN)NN)NNN)NN)rVrWrXrYr;r<rrrHrJrrKrIr rcrrrrfrrrrr!rrrr rrrrrrrr rr rr r7r rrzrpropertyrrrr staticmethodrrrrrrrrdrrrrrrrrrrr:rrrrLrMr rrrr!r"r#r rrr+rr&r,r'r-r(r%r*rrrrrrrrrrrrrrrrNrrrrrr6rrrrr rRrr(r*r/r5r7r9r8Z Optimizerrr?r@rCrErFrQr[rrobjectrar]rrre __classcell__r]r]r3r^rGsQ6 -     +   7 / %     "    = 9i   7 A   P  #     a     '  J  2   # "  'RO P &) # 5   E +   K   *  n   v 2F)rBrDr}csdd|D}t|dkr$tdStjkrHttdd|D}n"tjtfdd|D}|j |dj d|S) zCalculate gradient norm of an iterable of parameters. Returns: Total norm of the parameters (viewed as a single vector). cSsg|]}|jdk r|qSrr)r\r]r]r^rs z#_calc_grad_norm..rgcss |]}|jVqdSr)rriabsmaxrparr]r]r^rsz"_calc_grad_norm..cs&g|]}tjj|jtjdqS)r)rclinalg vector_normrriZfloat32rlrDr]r^rsr) rrcrrr1rkrnrorrerd)rBrDr4r]rpr^r0s     r0)r:rFr}cs&fdd}dd}i}t||||S)aN Constructs a mapping from flattened parameter (including non-FSDP-module parameters) to its unflattened parameter names. For non-FSDP-module parameters, these mapped-to lists always contain a single element. The unflattened parameter names should match the keys of the model state dict. For shared parameters, only the first parameter name is included (following the ``torch.nn.Module.parameters()`` order). Args: model (torch.nn.Module): Root module (which may or may not be a :class:`FullyShardedDataParallel` instance). dedup_shared_params (bool): If ``True``, only includes the first list of unflattened parameter names corresponding to a parameter in the module walk order; if ``False``, then includes all of the unflattened parameter names. csvt|tsr|jddD]Z\}}t|tkr0|jn|g}fdd|D}||k}|s^|||<qs|||qdS)NFrLcsg|]}t|qSr])rS)rrrJr]r^rszG_get_param_to_unflat_param_names..module_fn..)rrGrrr:r6Z_prefixed_param_namesrm)rrJrGrsrHZmodule_prefixed_param_namesZfully_prefixed_param_namesZis_shared_paramrErqr^rQ s   z3_get_param_to_unflat_param_names..module_fncSs|Srr])rGr]r]r^rS"sz3_get_param_to_unflat_param_names..return_fn)r/)r:rFrQrSrGr]rEr^rs r)r:r}cCsdt|}|D]<}t|dks(tdt|dkrtdt|d|qdd|D}|S)a( Constructs a mapping from parameters to their parameter names. ``model`` should not contain any :class:`FullyShardedDataParallel` instances, which means that none of the parameters should be ``FlatParameter`` s. As a result, compared to :meth:`_get_param_to_unflat_param_names`, the mapped values may be flattened from singleton :class:`list` s to the contained names themselves. Args: model (torch.nn.Module): Root module, which should not contain any :class:`FullyShardedDataParallel` instances. rzE`_get_param_to_unflat_param_names()` should not construct empty listsr"z=Each parameter should only map to one parameter name but got rgcSsi|]\}}||dqS)rr])rrHrr]r]r^rOCsz,_get_param_to_param_name..)rvaluesrr?rrU)r:Zparam_to_param_namesrrJr]r]r^rT+s  rTcCst|}tt||S)zCConstructs the inverse mapping of :meth:`_get_param_to_param_name`.)rTrcr|rrr)r:rJr]r]r^rYJsrY) tensor_namer}cCs4|tdd}|tdd}|tdd}|S)zPCleans the parameter or buffer name by removing any module wrapper prefixes.rUr)rrr<r)rsr]r]r^rSRs)T)rrrVrUrrr%rrZ dataclassesrenumrrtypingrrrr r r r r rrrrrrrrcZtorch.distributedr2rZ;torch.distributed.algorithms._checkpoint.checkpoint_wrapperZ algorithmsZ _checkpointrZtorch.nnr;Ztorch.nn.functionalZ functionalrZtorch.autogradrrZ'torch.distributed._shard.sharded_tensorrrrrZ(torch.distributed.algorithms._comm_hooksrrZ"torch.distributed.distributed_c10drZtorch.distributed.utilsrrr Ztorch.nn.parameterr!Z _optim_utilsr#r$r%r&r'r(r)r*r+r,Z_fsdp_extensionsr-r._utilsr/r0r1r2r3r4r5rr6r7r8r9r:Zflatten_params_wrapperr;r<r=rdr>r?r@rAr`Z torchdistxrBrC ImportErrorrar>Z_symbolic_tracerDrErF__all__rrrrnrHrIrJrKrRrLrMrNrOrPrmrnrorrQrruryrr\rZr[rr<rGr/rr0rfrrrTrYrSr]r]r]r^s  D       0 $   $6   ]z 4