o oi @sddlZddlZddlmZmZddlmZddlmZddl m Z ddl m Z m Z mZmZmZmZmZmZmZmZmZmZmZddlZddlmZddlmZdd lmZdd l m!Z!dd l"m#Z#dd l$m%Z%m&Z&dd l'm(Z(ddl)m*Z*m+Z+m,Z,ddl-m.Z.ddl/m0Z0ddl1m2Z2ddl3m4Z4ddl5m6Z6ddl7m8Z8m9Z9m:Z:m;Z;mZ>m?Z?m@Z@mAZAmBZBddl=mCZDddlEmFZFmGZGddlHmIZIddlJmKZKmLZLmMZMmNZNddlOmPZPmZmQZQddlRmSZSddlTmUZUmVZVe rddlWmXZXddlYmZZZm[Z[m\Z\ddl]m^Z^eeee!ee!e_e`ge_fe^fZaee\ed fZbd!Zcejdd"eed#d$Gd%d&d&e4e;Zfd'eeee!eee!fd(ed)d*efd+d,Zgd-ed)d.ed*efd/d0Zhd1e!d2ed*dfd3d4ZiGd5d6d6e:Zjd7eee_d8fd*d8fd9d:Zkd;dd?Zld@e#d*e_fdAdBZmd1e!d*edCfdDdEZn Fd[d1e!dGe`dHe_d*edCfdIdJZodKe d*e_fdLdMZpdKe d*e_fdNdOZqd1erd*e%e!fdPdQZsd1ejtj!dRejud*dfdSdTZvdUeewe fdKe d*dfdVdWZxdXeewe fdKe d*dfdYdZZydS)\N) ExitStack nullcontext) timedelta)partial)Path) TYPE_CHECKINGAnyCallableContextManagerDict GeneratorListLiteralOptionalSetTupleTypeUnion)RequirementCache)rank_zero_only)Tensor)Module) Optimizer) TypeGuardoverride) Accelerator) CheckpointIOClusterEnvironment Precision)default_pg_timeout) FSDPPrecision)_SubprocessScriptLauncher)ParallelStrategy)_StrategyRegistry) TBroadcast _apply_filter_BackwardSyncControl_Sharded!_validate_keys_for_strict_loading)ReduceOp_distributed_is_initialized-_get_default_process_group_backend_for_device_init_dist_connection_sync_ddp_if_availablegroup)_TORCH_GREATER_EQUAL_2_2_TORCH_GREATER_EQUAL_2_3)&_has_meta_device_parameters_or_buffers)_METADATA_FILENAME _lazy_load_materialize_tensors_move_state_into)rank_zero_deprecationrrank_zero_warn) reset_seed)_PATH _Stateful) DeviceMesh) CPUOffloadMixedPrecisionShardingStrategyModuleWrapPolicy) FULL_SHARD SHARD_GRAD_OPNO_SHARD HYBRID_SHARD)fsdpfsdp_cpu_offloadignorez.*FSDP.state_dict_type.*)categorymessagec!s&eZdZdZdddddeddddddddfdeedeeej dee dee d ee d ee d eed dfd eddeddeeeeeeefdeddddeddeeeedfdeddf fdd ZeedefddZejededdfd dZeedej fd!d"Zedefd#d$Zejd%eddfd&d$Zedefd'd(Zeedee effd)d*Z edee fd+d,Z!ededfd-d.Z"eede#fd/d0Z$e$jedee#ddfd1d0Z$edmd2d3Z%edmfd4d5 Z&ed6ed7ee'deeee'ffd8d9Z(ed6edefd:d;Z)ed Z*ed6eddfd?d@Z+edndAeede,fdBdCZ-ede,fdDdEZ.e FdodGe/dHeedIeee0e fde/fdJdKZ1edLededdfdMdNZ2edpdPe3dQede3fdRdSZ4e T Udqd6edde fdidjZ?dmdkdlZ@ZAS)t FSDPStrategyae Strategy for Fully Sharded Data Parallel provided by torch.distributed. Fully Sharded Training shards the entire model across all available GPUs, allowing you to scale model size, whilst using efficient communication to reduce overhead. In practice, this means we can remain at parity with PyTorch DDP, whilst scaling our model sizes dramatically. The technique is similar to ZeRO-Stage 3. For more information check out `this blogpost `__. Defaults have been set and options have been exposed, but may require configuration based on your level of memory/speed efficiency. We suggest having a look at `this tutorial `__ for more information. Arguments: cpu_offload: See ``cpu_offload`` parameter in :class:`torch.distributed.fsdp.FullyShardedDataParallel`. mixed_precision: See ``mixed_precision`` parameter in :class:`torch.distributed.fsdp.FullyShardedDataParallel`. auto_wrap_policy: Same as ``auto_wrap_policy`` parameter in :class:`torch.distributed.fsdp.FullyShardedDataParallel`. For convenience, this also accepts a set of the layer classes to wrap. activation_checkpointing: Deprecated. Use ``activation_checkpointing_policy``. activation_checkpointing_policy: Same as ``auto_wrap_policy`` parameter in :class:`torch.distributed.fsdp.FullyShardedDataParallel` but used when selecting the modules for which you want to enable activation checkpointing. Enabling this can free up a significant amount of memory at the cost of speed since activations in these layers need to be recomputed during backpropagation. For convenience, this also accepts a set of the layer classes to wrap. sharding_strategy: Select whether to shard model parameters, gradients, optimizer states, or a combination of them. Available values are: - ``"FULL_SHARD"``: Shards model parameters, gradients, and optimizer states (default). - ``"SHARD_GRAD_OP"``: Shards gradients and optimizer states only. Model parameters get replicated. - ``"NO_SHARD"``: No sharding (identical to regular DDP). - ``"HYBRID_SHARD"``: Shards model parameters, gradients, and optimizer states within a single machine, but replicates across machines. See also the `device_mesh` parameter below. Also accepts a :class:`torch.distributed.fsdp.ShardingStrategy` enum value. device_mesh: A tuple `(replication size, sharding size)` that defines over how many devices to shard and replicate the model. The product of the two numbers must equal the world size. Only valid in combination with the `HYBRID_SHARD` sharding strategy. state_dict_type: The format in which the state of the model and optimizers gets saved into the checkpoint. - ``"full"``: The full weights and optimizer states get assembled on rank 0 and saved to a single file. - ``"sharded"``: Each rank saves its shard of weights and optimizer states to a file. The checkpoint is a folder with as many files as the world size. \**kwargs: See available parameters in :class:`torch.distributed.fsdp.FullyShardedDataParallel`. NrBsharded acceleratorparallel_devicescluster_environment precisionprocess_group_backendtimeout cpu_offloadr=mixed_precisionr>auto_wrap_policy_POLICYactivation_checkpointingactivation_checkpointing_policysharding_strategy_SHARDING_STRATEGYstate_dict_type)fullrL device_meshr<kwargsreturnc stj||||dd|_||_||_t|_t| ||_|j dd|dur3t s.t d||jd<t | | |_ | |_ t| |j|_t||_||_dS)N)rMrNrOrPuse_orig_paramsTz=The `device_mesh` argument is only supported in torch >= 2.2.r])super__init__ _num_nodes_process_group_backend_timeout_FSDPBackwardSyncControl_backward_sync_control_auto_wrap_policy_kwargs _fsdp_kwargs setdefaultr0 ValueError _activation_checkpointing_kwargs_state_dict_type_init_sharding_strategyrY_init_cpu_offloadrSrT)selfrMrNrOrPrQrRrSrTrUrWrXrYr[r]r^ __class__T/home/ubuntu/.local/lib/python3.10/site-packages/lightning/fabric/strategies/fsdp.pyrcs.    zFSDPStrategy.__init__cCtdt|jd)NThe `z3` does not use the `CheckpointIO` plugin interface.NotImplementedErrortype__name__rqrtrtru checkpoint_iozFSDPStrategy.checkpoint_ioiocCrv)Nrwz3` does not support setting a `CheckpointIO` plugin.rx)rqrrtrtrur}r~cCs|jdusJ|j|jSN)rN local_rankr|rtrtru root_devices zFSDPStrategy.root_devicecC|jSrrdr|rtrtru num_nodeszFSDPStrategy.num_nodesrcCs ||_dSrr)rqrrtrtrurs cCs|jdur t|jSdS)Nr)rNlenr|rtrtru num_processesszFSDPStrategy.num_processescCs|j|j|jdS)N) num_replicasrank)rr global_rankr|rtrtrudistributed_sampler_kwargssz'FSDPStrategy.distributed_sampler_kwargscCrr)rer|rtrtrurQrz"FSDPStrategy.process_group_backendcCs&|jr|jS|j}t|tr|jSdSr)rTrP isinstancer mixed_precision_configrqpluginrtrtrurs  z#FSDPStrategy.mixed_precision_configcCs(|j}|durt|tsJ|StdS)Nz32-true) _precisionrr rrtrtrurPs zFSDPStrategy.precisioncCs*|durt|tstd|||_dS)NzGThe FSDP strategy can only work with the `FSDPPrecision` plugin, found )rr TypeErrorr)rqrPrtrtrurPs cCs2|jdusJ|jjst|j|j|j|_dSdSr)rOcreates_processes_externallyr!rr _launcherr|rtrtru_configure_launchersz FSDPStrategy._configure_launchercsNt|t|jdtr%ddlm}|d|jd|jd<dSdS)Nr]r)init_device_meshcuda) rbsetup_environment_setup_distributedrrjgettupletorch.distributed.device_meshr)rqrrrrtrurs  zFSDPStrategy.setup_environmentmodule optimizerscCs<|jd}|durtdt|jd||}||fS)zWraps the model into a :class:`~torch.distributed.fsdp.fully_sharded_data_parallel.FullyShardedDataParallel` module and sets `use_orig_params=True` to keep the reference to the original parameters in the optimizer.raFz You set `z(use_orig_params=False)` but this is not supported when setting the model and optimizer up jointly. Either set it to `True` or set the objects up in this order: Create the model, call `setup_module`, create the optimizer, call `setup_optimizer`.)rjrrlrzr{ setup_module)rqrrrartrtrusetup_module_and_optimizers s  z(FSDPStrategy.setup_module_and_optimizerscsddlmtfdd|Dr)t|rtdd|jvr(td|jd=nd ||j|j|j |j j d|j}t ||j t ||j|S) z|Wraps the model into a :class:`~torch.distributed.fsdp.fully_sharded_data_parallel.FullyShardedDataParallel` module.rFullyShardedDataParallelc3|]}t|VqdSrr.0modrrtru #z,FSDPStrategy.setup_module..zYThe model is already wrapped in `FSDP` but there are still parameters on the meta device.rUz_A FSDP `auto_wrap_policy` is set, but the model is already wrapped. The policy will be ignored.)rrSrTrY device_idNrt)torch.distributed.fsdpranymodulesr2r8rjrSrrYrindex_move_torchmetrics_to_device_setup_activation_checkpointingrmrqrrtrrurs0   zFSDPStrategy.setup_module optimizercs,|jdr t|St|std|S)aISet up an optimizer for a model wrapped with FSDP. This setup method doesn't modify the optimizer or wrap the optimizer. The only thing it currently does is verify that the optimizer was created after the model was wrapped with :meth:`setup_module` with a reference to the flattened parameters. razThe optimizer does not seem to reference any FSDP parameters. HINT: Make sure to create the optimizer after setting up the model.)rjrrbsetup_optimizer_optimizer_has_flat_paramsrl)rqrrrrtrur?s zFSDPStrategy.setup_optimizercCsdSrrtrrtrtrumodule_to_deviceRszFSDPStrategy.module_to_device empty_initcCsD|j}|}t}|r|td|||||S)Nmeta)rPmodule_init_contextmodule_sharded_contextr enter_contexttorchdevice)rqrprecision_init_ctxmodule_sharded_ctxstackrtrtrurVs   z FSDPStrategy.module_init_contextcCs>ddlm}ddlm}|d||j|j|j|jjd|j S)Nrr) enable_wrap) wrapper_clsrSrTrYrrt) 2torch.distributed.fsdp.fully_sharded_data_parallelrtorch.distributed.fsdp.wraprrSrrYrrrj)rqrrrtrtrurds  z#FSDPStrategy.module_sharded_contextmeantensorr/ reduce_opcCst|tr t|||dS|S)N)r)rrr-)rqrr/rrtrtru all_reducers zFSDPStrategy.all_reduceargscOs>tsdStjdkrtjj|jjgddStjdS)Nnccl) device_ids)r*r distributed get_backendbarrierrr)rqrr^rtrtrurzs zFSDPStrategy.barrierrobjsrccCs,ts|S|g}tjj||tjd|dS)Nr.r)r*rrbroadcast_object_list_groupWORLD)rqrrrtrtru broadcasts zFSDPStrategy.broadcast@Tmax_norm norm_typeerror_if_nonfinitecCsPddlm}t||stdt|jd|jjd|j||j ||dS)zClip gradients by norm.rrzFGradient clipping with FSDP is only possible if the module passed to `zE.clip_gradients_norm` is wrapped in `FullyShardedDataParallel`. Got: .)rr) rrrrrzr{rsrPunscale_gradientsclip_grad_norm_)rqrrrrrrrtrtruclip_gradients_norms  z FSDPStrategy.clip_gradients_normpathstatestorage_optionsfiltercCs|durtd|dur|jdkrtdt||}|r0|jdkr0t|s0td|ddlm }d d | D}t |dkrIt d t |d krSt d |d}|jdkr| rd||jdddt|}i} i} |A|D]4\} } t| tr| } | }nt| tr||| } | }n t| tr| n| } | }t| |pi| |qzWdn1swYt| ||jdkrt| |tdSdS|jdkrFt|rt|t ||j!d}i}|?|D]2\} } t| tr| } nt| tr||| } n t| tr| n| } t| |p!i| |qWdn 1s1wY|jdkrDt||dSdSt d|j)aSave model, optimizer, and other state to a checkpoint on disk. If the state-dict-type is ``'full'``, the checkpoint will be written to a single file containing the weights, optimizer state and other metadata. If the state-dict-type is ``'sharded'``, the checkpoint gets saved as a directory containing one file per process, with model- and optimizer shards stored per file. Additionally, it creates a metadata file `meta.pt` with the rest of the user's state (only saved from rank 0). Nz`FSDPStrategy.save_checkpoint(..., storage_options=...)` is not supported because `FSDPStrategy` does not use the `CheckpointIO`.rLzVFSDP doesn't support loading sharded filtered checkpoints, so saving them is disabled.r\z/The checkpoint path exists and is a directory: rrcSsg|]}t|r|qSrt_has_fsdp_modules)rrrtrtru sz0FSDPStrategy.save_checkpoint..aCould not find a FSDP model in the provided checkpoint state. Please provide the model as part of the state like so: `save_checkpoint(..., state={'model': model, ...})`. Make sure you set up the model (and optimizers if any) through the strategy before saving the checkpoint.r`zFound multiple FSDP models in the given state. Saving checkpoints with FSDP is currently limited to a single model per checkpoint. To save multiple models, call the save method for each model separately with a different path.T)parentsexist_ok) world_sizezUnknown state_dict_type: )"rrnryrris_dir_is_sharded_checkpointIsADirectoryErrorrrvaluesrrlis_fileunlinkmkdir_get_sharded_state_dict_contextitemsrr state_dictroptim_state_dictr;r%_distributed_checkpoint_saverrsaver3shutilrmtree_get_full_state_dict_contextr)rqrrrrFSDPrrstate_dict_ctxconverted_statemetadatakeyr converted target_dict full_statertrtrusave_checkpoints               zFSDPStrategy.save_checkpointstrictc Cs|s td|dt||}t|tr'ddlm}||||j|diSt|tr0t dddl m }ddl m }d d |D}t|dkrOtd d d |D}t|d krbtdt|d\} } t|rt| } | J| | i} t| || j| | |d|rddlm} | |d}|D]\}}|| | ||d}|j||| |d}||qWdn1swYt|t}|||}t|||d|D]}||vrq||||<q|St|rvt |}ddlm!}m"}||| | |j|dt|triSt#|}|D]3\}}t$| |jdd|||| }|j|| |d}||Wdn 1sQwYq$|||}t|||dt%|||d|Stdt&|d)zOLoad the contents from a checkpoint and restore the state of the given objects.z,Got FSDPStrategy.load_checkpoint(..., state=z) but a state with at least a model instance to reload is required. Pass it in like so: FSDPStrategy.load_checkpoint(..., state={'model': model, ...})r) _load_raw_module_state_from_path)rrrz`Loading a single optimizer object from a checkpoint is not supported yet with the FSDP strategy.)!load_sharded_optimizer_state_dictrcSsi|] \}}t|r||qSrtr)rrrrtrtru %sz0FSDPStrategy.load_checkpoint..aCould not find a FSDP model in the provided checkpoint state. Please provide the model as part of the state like so: `load_checkpoint(..., state={'model': model, ...})`. Make sure you set up the model (and optimizers if any) through the strategy before loading the checkpoint.cSs i|] \}}t|tr||qSrt)rr)rroptimrtrtrur,s r`zFound multiple FSDP models in the given state. Loading checkpoints with FSDP is currently limited to a single model per checkpoint. To load multiple models, call the load method for each model separately with a different path.)rFileSystemReaderr)model_state_dict optimizer_keystorage_reader)rmodelrN)_load_raw_module_state _rekey_optimizer_state_if_neededF)r rank0_only)source destinationkeysz The path z does not point to a valid checkpoint. Make sure the path points to either a directory with FSDP checkpoint shards, or a single file with a full checkpoint.)'rlrrrr*lightning.fabric.strategies.model_parallelrrrry&torch.distributed.checkpoint.optimizerrrrrrlistrrr_distributed_checkpoint_loadload_state_dicttorch.distributed.checkpointroptim_state_dict_to_loadrloadr3rr(pop_is_full_checkpointr4r r r5rr6str)rqrrrrrrrr module_keyrr module_staterreader optim_keyr optim_state flattened_osdrrequested_metadata_keysr checkpointr r temp_state_dictrrtrtruload_checkpoints                 zFSDPStrategy.load_checkpointstrategy_registrycCs4tjsdS|jd|dd|jd|ddddS)NrFz+Fully Sharded Data Parallel (FSDP) training) descriptionrGzQFully Sharded Data Parallel (FSDP) training with Full Sharding and CPU OffloadingT)r'rS)rr is_availableregister)clsr&rtrtruregister_strategiess  z FSDPStrategy.register_strategiescCs>t|||_|jdusJt|j|j|jddS)N)rR)r9_set_world_ranks_get_process_group_backendrerOr,rfr|rtrtrurs  zFSDPStrategy._setup_distributedcCs|jpt|jSr)rer+rr|rtrtrur-sz'FSDPStrategy._get_process_group_backendcCsJ|jdur|j|j|j|j|j|j|j|jt_ t _ dSr) rOset_global_rank node_rankrrset_world_sizerrrrutils_rank_zero_onlyr|rtrtrur,s zFSDPStrategy._set_world_ranks)r_Nr)Nr)r)rT)NN)NT)Br{ __module__ __qualname____doc__rrrr rrrrrrrboolrrrrintrrcpropertyrrr}setterrrrr rrQrr rPrrrrrrrr rrrr)rrr$rfloatrr:r rr% classmethodr#r+rr-r, __classcell__rtrtrrrurK[sF5       .   !       c"   rKrWrXrVr_cs|dur |dur iS|dur|durtd|durs z2_activation_checkpointing_kwargs..rU)rlrrrr7setri)rWrXrtr>rurms&    rmpolicyr^cCs6|dur|St|trddlm}||}||d<|S)Nrr@rU)rrArrA)rBr^rArtrtruris  riractivation_checkpointing_kwargscsx|sdSddlmtfdd|DrtddSddlm}m}m}ts0t ||j d}||fd|i|dS) NrCheckpointWrapperc3rrrrrDrtrurrz2_setup_activation_checkpointing..ztFSDP checkpointing is configured, but the model already contains checkpointed layers. Checkpointing will be ignored.)CheckpointImplapply_activation_checkpointingcheckpoint_wrapper)checkpoint_implcheckpoint_wrapper_fn) ;torch.distributed.algorithms._checkpoint.checkpoint_wrapperrErrr8rFrGrHr0r NO_REENTRANT)rrCrFrGrHrtrDrurs rc@s&eZdZedededefddZdS)rgrenabledr_cCsH|stSddlm}t||s tdt|jd|jjd|S)zuBlocks gradient synchronization inside the :class:`~torch.distributed.fsdp.FullyShardedDataParallel` wrapper.rrzABlocking backward sync is only possible if the module passed to `zB.no_backward_sync` is wrapped in `FullyShardedDataParallel`. Got: r) rrrrrrzr{rsno_sync)rqrrMrrtrtruno_backward_syncs  z)_FSDPBackwardSyncControl.no_backward_syncN)r{r2r3rrr5r rOrtrtrtrurgsrgrSr=cCs(ddlm}t||r |S|t|dS)Nr)r=)offload_params)rr=rr5)rSr=rtrtrurps rprYrZr?cCsddlm}|ddur|ddurtdt|tr#||n|}d|jvrC|ddurC|ddurC|ddurCtd|S) Nr)r? process_groupr]znThe arguments `FSDPStrategy(process_group=..., device_mesh=...)` are mutually exclusive.Pass only one of them.HYBRIDrUzThe hybrid sharding strategy requires you to pass at least one of the parameters: `auto_wrap_policy`, `process_group` tuple, or `device_mesh`.) rr?rrlrruppername RuntimeError)rYr^r?strategyrtrtruros  rorcCstdd|jDS)Ncss*|]}|dD] }t|ddVqqdS)params_fsdp_flattenedFN)getattr)rr/paramrtrtrurs   z-_optimizer_has_flat_params..)r param_groups)rrtrtrursr)NNNcCsLddlm}ddlm}m}m}|dd}|dd}|j||j||d}|S)Nrr)ShardedOptimStateDictConfigShardedStateDictConfig StateDictTypeT)offload_to_cpurr[state_dict_configoptim_state_dict_config)rrtorch.distributed.fsdp.apir\r]r^r[SHARDED_STATE_DICT)rrr\r]r^rarbstate_dict_type_contextrtrtrur s   rTrr c CsXddlm}m}ddlm}ddlm}|d|d}|d|d}|j||j||d} | S)Nr)FullStateDictConfigr^r)FullOptimStateDictConfigT)r_r r`)rrfr^rrcrgr[FULL_STATE_DICT) rrr rfr^rrgrarbrertrtrur/s    rrcCs|o |tS)z]A heuristic check to determine whether the path points to a directory with checkpoint shards.)rr3rrrtrtrurBsrcCs|Sr)rrrtrtrurGsrcs0ddlmt|totfdd|DS)Nrrc3rrrrmrrtrurNrz$_has_fsdp_modules..)rrrrrr)rrtrrurKs $rrcsBtdsdSddlmfdd|DD]}||qdS)N torchmetricsrMetricc3s|] }t|r|VqdSrrrirlrtrurYsz/_move_torchmetrics_to_device..)rrkrmrto)rrmetricrtrlrurQs   rrcCsdtrddlm}|||ddSddlm}trddlm}nddlm}||dd}|||dS)Nr)r checkpoint_id)FileSystemWriter)save_state_dictT)rsingle_file_per_rank)r1rrrrr0rs)rrrrrwriterrtrtrur]s    rrcCsbtrddlm}|||ddSddlm}trddlm}nddlm}||d}|||dS)Nr)rrpr)rr)r1rrrr0r)rrrrrrtrtrurps    r)T)zrwarnings contextlibrrdatetimer functoolsrpathlibrtypingrrr r r r r rrrrrrr lightning_utilities.core.importsr"lightning_utilities.core.rank_zerorr1rtorch.nnr torch.optimrtyping_extensionsrrlightning.fabric.acceleratorsrlightning.fabric.pluginsrrr5lightning.fabric.plugins.collectives.torch_collectiver'lightning.fabric.plugins.precision.fsdpr 7lightning.fabric.strategies.launchers.subprocess_scriptr!$lightning.fabric.strategies.parallelr"$lightning.fabric.strategies.registryr#$lightning.fabric.strategies.strategyr$r%r&r'r(&lightning.fabric.utilities.distributedr)r*r+r,r-r/r"lightning.fabric.utilities.importsr0r1lightning.fabric.utilities.initr2lightning.fabric.utilities.loadr3r4r5r6$lightning.fabric.utilities.rank_zeror7r8lightning.fabric.utilities.seedr9 lightning.fabric.utilities.typesr:r;rr<rr=r>r?rrAr5r6rVrZ _FSDP_ALIASESfilterwarnings FutureWarningrKrmrirrgrprorrrrrobjectrnnrrrrrrtrtrtrus    <                $V    "