o ziw@sddlZddlZddlmZmZddlmZddlmZddl m Z m Z m Z m Z mZmZmZmZmZmZmZmZmZddlZddlmZddlmZddlmZdd lmZdd l m!Z!ddl"Z#dd l$m%Z%m&Z&dd l'm(Z(dd l)m*Z*ddl+m,Z,m-Z-m.Z.m/Z/m0Z0m1Z1m2Z2m3Z3m4Z4m5Z5m6Z6m7Z7m8Z8m9Z9ddl:m;Z;ddlZ>m?Z?m@Z@ddlede#eeffdbdcZ?edzdee@eefdfeddfdgdhZAedieBde#ee6ffdjdkZCedee@eefddfdldmZDe dwdee#eefdneEdoe eddffdpdq ZFedreEde#eeffdsdtZGZHS){ 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`. fsdp_registered_strategiesNrHfull acceleratorzpl.accelerators.Acceleratorparallel_devicescluster_environment checkpoint_ioprecision_pluginprocess_group_backendtimeout cpu_offloadrDmixed_precisionrEauto_wrap_policy_POLICYactivation_checkpointingactivation_checkpointing_policysharding_strategy_SHARDING_STRATEGYstate_dict_type)rOsharded device_meshrCkwargsreturnc stj|||||dd|_||_||_t||_| |_t| ||_ |dur1t s,t d||j d<t | |j |_ |j ddt| | |_||_dS)N)rPrQrRrSrTz=The `device_mesh` argument is only supported in torch >= 2.2.rause_orig_paramsT)super__init__ num_nodes_process_group_backend_timeoutr#rWrXrrbr1 ValueErrorr$r] setdefaultr_state_dict_type)selfrPrQrRrSrTrUrVrWrXrYr[r\r]r_rarb __class__U/home/ubuntu/.local/lib/python3.10/site-packages/pytorch_lightning/strategies/fsdp.pyrgs.    zFSDPStrategy.__init__cCs|jdusJ|j|jSN)rQ local_rankrnrqrqrr root_devices zFSDPStrategy.root_devicecCs|jdur t|jSdS)Nr)rQlenrurqrqrr num_processesszFSDPStrategy.num_processescC|jSrs)rirurqrqrrrUz"FSDPStrategy.process_group_backendcCs&|jr|jS|j}t|tr|jSdSrs)rXrT isinstancer;mixed_precision_configrnpluginrqrqrrr|s  z#FSDPStrategy.mixed_precision_configcCs(|j}|durt|tsJ|StdS)Nz32-true)_precision_pluginr{r;r}rqrqrrrTs zFSDPStrategy.precision_plugincCs*|durt|tstd|||_dS)NzGThe FSDP strategy can only work with the `FSDPPrecision` plugin, found )r{r; TypeErrorr)rnrTrqrqrrrTs  cCs|j|j|jdS)N) num_replicasrank)rhrx global_rankrurqrqrrdistributed_sampler_kwargssz'FSDPStrategy.distributed_sampler_kwargscCdS)NTrqrurqrqrrrestore_checkpoint_after_setupz+FSDPStrategy.restore_checkpoint_after_setupcCr)NFrqrurqrqrrlightning_restore_optimizerrz(FSDPStrategy.lightning_restore_optimizercstt|jjdt|||_ |j dus"Jt |j |j |j dt |jdtrHddlm}|d|jd|jd<dSdS)Nz: setting up distributed...)rVrar)init_device_meshcuda)rfsetup_environmentlogdebugrp__name__r6set_world_ranks_get_process_group_backendrirRr-rjr{rbgettupletorch.distributed.device_meshr)rnrrorqrrrs   zFSDPStrategy.setup_environmentcCs|jpt|jSrs)rir,rvrurqrqrrrsz'FSDPStrategy._get_process_group_backendcCsJ|jdur|j|j|j|j|j|j|j|jt_ t _ dSrs) rRset_global_rank node_rankrxrtset_world_sizerhrrrutils_rank_zero_onlyrurqrqrrrs zFSDPStrategy.set_world_rankscCs2|jdusJ|jjst|j|j|j|_dSdSrs)rRcreates_processes_externallyr<rxrh _launcherrurqrqrr_configure_launcher sz FSDPStrategy._configure_launchermodelcsddlmtfdd|Dr)t|rtdd|jvr(td|jd=n!td|j j d |jd ||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.rFullyShardedDataParallelc3s|]}t|VqdSrs)r{).0modrrqrr ,z,FSDPStrategy._setup_model..zYThe model is already wrapped in `FSDP` but there are still parameters on the meta device.rYz_A FSDP `auto_wrap_policy` is set, but the model is already wrapped. The policy will be ignored.z&setting up FSDP model with device id: z , kwargs: )modulerWrXr] device_idNrq)torch.distributed.fsdpranymodulesr2rBrbrrrvindexrWr|r]r'r)r)rnrrqrrr _setup_model&s2   zFSDPStrategy._setup_modeltrainer pl.TrainercCs|jdusJ|j||jdusJ|jjtjkr&|jr&|j|j|_|j |j|_t d|j r9t dn||j|_||jjtjkrP||||jjtjkrdt|j|jdSdS)Nconfigure_sharded_modelzYou have overridden `LightningModule.configure_sharded_model` hook. It will assume that all the layers are already wrapped for sharding and won't wrap the entire model using `FullyShardedDataParallel`.)rPsetuprstatefnr?FITTING _layer_syncapplyrTconvert_moduler@lightning_modulerArbarriersetup_optimizerssetup_precision_pluginr5 optimizersrv)rnrrqrqrrrIs$   zFSDPStrategy.setupc s||jdrt|Sd}zt|Wnty5}zdt|vr)d}WYd}~nd}~ww|sBtdd|jDrFtddS)NreFz%optimizer got an empty parameter listTcss|]}t| VqdSrs)r()r optimizerrqrqrrrxrz0FSDPStrategy.setup_optimizers..zThe optimizer does not seem to reference any FSDP parameters. HINT: Make sure to create the optimizer after setting up the model by referencing `self.trainer.model.parameters()` in the `configure_optimizers()` hook.) _reset_optimizers_and_schedulersrbrrfrrkstrrr)rnrinvalid_params_errorexrorqrrrds"   zFSDPStrategy.setup_optimizerscCdSrsrqrurqrqrrmodel_to_devicerzFSDPStrategy.model_to_device empty_init)NNNc cs|rtdnt}|)|j dVWdn1s!wYWddSWddS1s9wYdS)Nmeta)torchdevicerrTtensor_init_context)rnrempty_init_contextrqrqrrrs Pz FSDPStrategy.tensor_init_contextccst|jjdddlm}ddlm}|d||j|j |j |j j d|j  dVWddS1s9wYdS)Nz : entered model_sharded_context.rr) enable_wrap) wrapper_clsrWrXr]rrq)rrrpr2torch.distributed.fsdp.fully_sharded_data_parallelrtorch.distributed.fsdp.wraprrWr|r]rvrrb)rnrrrqrqrrmodel_sharded_contexts  "z"FSDPStrategy.model_sharded_contextnamecCs<tsdStjdkrtjj|ddStjdS)Nnccl) device_ids)r+r distributed get_backendr_determine_device_ids)rnrrqrqrrrs zFSDPStrategy.barrierrobjsrccCs,ts|S|g}tjj||tjd|dS)Nr/r)r+rrbroadcast_object_list_groupWORLD)rnrrrqrqrr broadcasts zFSDPStrategy.broadcastmeantensorr0 reduce_opcCst|tr t|||dS|S)aReduces a tensor from several distributed processes to one aggregated tensor. Args: tensor: the tensor to sync and reduce group: the process group to gather results from. Defaults to all processes (world) reduce_op: the reduction operation. Defaults to 'mean'/'avg'. Can also be a string 'sum' to calculate the sum during reduction. Return: reduced value, except when the input was not a tensor the output remains is unchanged )r)r{rr.)rnrr0rrqrqrrreduces zFSDPStrategy.reducecCs |jjgSrs)rvrrurqrqrrrs z"FSDPStrategy._determine_device_idscCst|jjd|j}|dur0|jdur0|jjjtj kr0|j r0|j dus(J|j |j |_ |j dus7J|jdus>J|j |j|jdS)Nz: tearing down strategy...)rrrprr_trainerrrr?rrrrevertrRrPteardownrT)rn pl_modulerqrqrrrs   zFSDPStrategy.teardowncCryrs)rN)clsrqrqrrget_registered_strategiesrzz&FSDPStrategy.get_registered_strategiesstrategy_registrycCsLtjsdS|jd|dd|jd|jd|ddd|jddS)NrMz+Fully Sharded Data Parallel (FSDP) training) descriptionfsdp_cpu_offloadzQFully Sharded Data Parallel (FSDP) training with Full Sharding and CPU OffloadingT)rrW)rr is_availableregisterrNappend)rrrqrqrrregister_strategiess  z FSDPStrategy.register_strategiescCs|jdusJ|jdkrt|j}n|jdkr t|j|jd}ntd|j| |jWdS1s:wYdS)Nr`rO world_sizeUnknown state_dict_type: )rrmr"r!rrk state_dict)rnstate_dict_ctxrqrqrrlightning_module_state_dicts   $z(FSDPStrategy.lightning_module_state_dictT checkpointstrictcCrrsrq)rnrrrqrqrrload_model_state_dictrz"FSDPStrategy.load_model_state_dictrcCsddlm}ddlm}t|tr|j}|jdusJ|jdkr=t|j| |j|WdS1s7wYn4|jdkrqt |j|j d| |j|}|j dkr`| ||j|j}|WdS1slwYtd|j)NrrOptimStateKeyTyper`rOrr)rrrr{r9 _optimizerrrmr"optim_state_dictr!rrrekey_optim_state_dictPARAM_IDrk)rnrFSDPrrrqrqrroptimizer_state s"      "   zFSDPStrategy.optimizer_statecCrrsrq)rnrrqrqrrload_optimizer_state_dict#rz&FSDPStrategy.load_optimizer_state_dictfilepathstorage_optionscs|durtdt||}|r#|jdkr#t|s#td||jdkrd|r0||j dddd| di}| d d t | d gDt |||jd krbt||tdSdS|jdkrzt|rrt|tj||d Std|j)Nz`FSDPStrategy.save_checkpoint(..., storage_options=...)` is not supported because `FSDPStrategy` does not use the `CheckpointIO`.rOz/The checkpoint path exists and is a directory: r`T)parentsexist_okrrcSsi|] \}}d||qS) optimizer_rq)ridx optim_staterqrqrr <s z0FSDPStrategy.save_checkpoint..optimizer_statesr)rrr)rrris_dirrmr&IsADirectoryErroris_fileunlinkmkdirpopupdate enumerater rrsavershutilrmtreerfsave_checkpointrk)rnrrrpathconverted_staterorqrrr (s0      zFSDPStrategy.save_checkpointcheckpoint_pathcCst||}ddlm}|jdusJ|jdusJt|rddlm}t |j}|_d|j i}t |||jj |d|jj d|jjjjtjkr|jrddlm}||d}t|jD]"\} } d| } ||d| |d } |j| | |j| d } | | qaWdn1swYt|t}|St|r?t|}t|d |j|j|jj d t |}ddlm}dd lm!}|"d}|dus|jjjjtjkr|St#|jt#|krt$dt#|jdt#|dt%|j|jdd:t&|j|D]*\}}t't(|d)dt*r|+||j,|j}|j||j|d }| |qWd|S1s8wY|St-dt.|d)Nrr)!load_sharded_optimizer_state_dictr)r)FileSystemReader)rr)model_state_dict optimizer_keystorage_reader)rroptimr)rrrrrzYou have configured z( optimizers but the checkpoint contains z optimizers to load. Please resume training with the same number of optimizers or edit the checkpoint manually to remove states.F)r rank0_onlyrz 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.)/rrrrrrr&&torch.distributed.checkpoint.optimizerrr"rrload_state_dictstrict_loadingrrrr?rrtorch.distributed.checkpointrr optim_state_dict_to_loadrloadrr%r3r*rrr4rrrw RuntimeErrorr!zipr{listkeysintr PARAM_NAMErkr)rnrrrrr module_staterreaderrr optim_keyr flattened_osdmetadatarrrr opt_staterqrqrrload_checkpointLs               zFSDPStrategy.load_checkpoint)rcN)rrrcNrs)r)Nr)T)Ir __module__ __qualname____doc__ strategy_namerNr r__annotations__rrrrrrr:rrboolrrr rr"rrgpropertyrrvrxrUr|r;rTsetterr rrrrrrrrrrrrr rrrr>rrr8rrr classmethodrrrrr rrrrr7r r* __classcell__rqrqrorrrLbs 3      1  "       " #$rL)tloggingr  contextlibrrdatetimerpathlibrtypingrrrr r r r r rrrrrr"lightning_utilities.core.rank_zerorrrtorch.nnr torch.optimrtyping_extensionsrpytorch_lightningpllightning_fabric.pluginsrr5lightning_fabric.plugins.collectives.torch_collectiverlightning_fabric.strategiesr lightning_fabric.strategies.fsdprrrrr r!r"r#r$r%r&r'r(r)*lightning_fabric.strategies.model_parallelr*&lightning_fabric.utilities.distributedr+r,r-r.r0r"lightning_fabric.utilities.importsr1lightning_fabric.utilities.initr2lightning_fabric.utilities.loadr3r4$lightning_fabric.utilities.optimizerr5lightning_fabric.utilities.seedr6 lightning_fabric.utilities.typesr7r8 pytorch_lightning.core.optimizerr9#pytorch_lightning.plugins.precisionr:(pytorch_lightning.plugins.precision.fsdpr;8pytorch_lightning.strategies.launchers.subprocess_scriptr<%pytorch_lightning.strategies.parallelr=%pytorch_lightning.strategies.strategyr> pytorch_lightning.trainer.statesr?)pytorch_lightning.utilities.model_helpersr@%pytorch_lightning.utilities.rank_zerorArBrrCrrDrErFrrGr0r"rZr^ getLoggerrrrLrqrqrqrrsV   <       @                $