o oit @sddlZddlmZmZddlmZddlmZddlm Z m Z m Z m Z m Z mZmZmZmZmZmZmZddlZddlmZddlmZddlmZdd lmZdd lmZdd lm Z dd l!m"Z"dd l#m$Z$ddl%m&Z&ddl'm(Z(ddl)m*Z*m+Z+ddl,m-Z-ddl.m/Z/ddl0m1Z1m2Z2m3Z3m4Z4ddl5m6Z6ddl7m8Z8ddl9m:Z:m;Z;ddlZ>m?Z?e rddl@mAZAeeeZBeeBe eeCeDgeCffZEGddde*e3ZFdedde de fdd ZGdeBd!ed"e de def d#d$ZHdeeBde de fd%d&ZIGd'd(d(e2ZJdS))N) ExitStack nullcontext)partial)Path) TYPE_CHECKINGAnyCallableContextManagerDictListLiteralOptionalSetTupleTypeUnion)Tensor)Module) Optimizer) DataLoader)override) Accelerator)_XLA_AVAILABLE) XLAPrecision)XLAEnvironment)XLACheckpointIO)ParallelStrategy_StrategyRegistry) _apply_filter) _XLALauncher) TBroadcast_BackwardSyncControl_Sharded!_validate_keys_for_strict_loading)get_filesystem) _EmptyInit)rank_zero_onlyrank_zero_warn)_PATH OptimizableReduceOpMpDeviceLoadercsxeZdZdZ        dkdeedeeejdee dee d ee d ee d e d d ededdffdd ZeedejfddZedefddZeede fddZejedee ddfddZeede fddZejedee ddfddZeedeffdd Zeedeffdd Zeedeffd!d" Zeedeffd#d$ Zedld%d&Zedlfd'd( Zed)ed*ee de!eee ffd+d,Z"ed)edefd-d.Z#ed)eddfd/d0Z$dmd1eede%fd2d3Z&ede%fd4d5Z'ed6e(dd7fd8d9Z)ed:e de fd;d<Z*ed:e+dedefd=d>Z,e ? @dnd)ed:e dAe-e.efdBe-e.efdCede/f dDdEZ0ed)ed:e dFe-e.efddfdGdHZ1edodIe/dJeedKede/fdLdMZ2e dpdNe-e/efdJeedOee-e3e4fde/fdPdQZ5edmdRee4dSededdfdTdUZ6edqdWe7dXede7fdYdZZ8e  dpd[e9d\e:e4e-ee effd]eed^ee:e4e;e4egeffddf d_d`Ze  @drd[e9d\ee-ee e:e4e-ee efffdcede:e4effdddeZ?e@edfeAddfdgdhZBde:fdidjZCZDS)sXLAFSDPStrategyaStrategy for training multiple XLA devices using the :func:`torch_xla.distributed.xla_fully_sharded_data_parallel.XlaFullyShardedDataParallel` method. .. warning:: This is an :ref:`experimental ` feature. For more information check out https://github.com/pytorch/xla/blob/master/docs/fsdp.md Args: auto_wrap_policy: Same as ``auto_wrap_policy`` parameter in :class:`torch_xla.distributed.fsdp.XlaFullyShardedDataParallel`. For convenience, this also accepts a set of the layer classes to wrap. activation_checkpointing_policy: 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. This accepts a set of the layer classes to wrap. 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 files for each shard in the host. Note that TPU VM multihost does not have a shared filesystem. sequential_save: With this enabled, individual ranks consecutively save their state dictionary shards, reducing peak system RAM usage, although it elongates the saving process. \**kwargs: See available parameters in :class:`torch_xla.distributed.fsdp.XlaFullyShardedDataParallel`. NshardedF acceleratorparallel_devices checkpoint_io precisionauto_wrap_policyactivation_checkpointing_policystate_dict_type)fullr.sequential_savekwargsreturnc  sXtsttttj||t||dt|_||_||_ | |_ ||_ ||_ d|_ dS)N)r/r0cluster_environmentr1r2F)rModuleNotFoundErrorstrsuper__init__r_XLAFSDPBackwardSyncControl_backward_sync_control_auto_wrap_policy _activation_checkpointing_policy _fsdp_kwargs_state_dict_type_sequential_save _launched) selfr/r0r1r2r3r4r5r7r8 __class__X/home/ubuntu/.local/lib/python3.10/site-packages/lightning/fabric/strategies/xla_fsdp.pyr>Ss   zXLAFSDPStrategy.__init__cCs(|jstdddlmm}|S)NzFAccessing the XLA device before processes have spawned is not allowed.r)rF RuntimeErrortorch_xla.core.xla_modelcore xla_model xla_device)rGxmrJrJrK root_deviceqszXLAFSDPStrategy.root_devicecCs|jdur t|jSdSNr)r0lenrGrJrJrK num_processeszszXLAFSDPStrategy.num_processescCs&|j}|durt|tsJ|StSN)_checkpoint_io isinstancerrGpluginrJrJrKr1~s zXLAFSDPStrategy.checkpoint_ioiocC*|durt|tstd|||_dS)NzHThe XLA strategy can only work with the `XLACheckpointIO` plugin, found )rYr TypeErrorrX)rGr\rJrJrKr1 cCs(|j}|durt|tsJ|StdS)Nz32-true) _precisionrYrrZrJrJrKr2s zXLAFSDPStrategy.precisioncCr])NzJThe XLA FSDP strategy can only work with the `XLAPrecision` plugin, found )rYrr^r`)rGr2rJrJrKr2r_c|jrtjSdSrS)rFr= global_rankrUrHrJrKrbzXLAFSDPStrategy.global_rankcrarS)rFr= local_rankrUrHrJrKrdrczXLAFSDPStrategy.local_rankcrarS)rFr= node_rankrUrHrJrKrerczXLAFSDPStrategy.node_rankcra)N)rFr= world_sizerUrHrJrKrgrczXLAFSDPStrategy.world_sizecCst||_dSrW)r _launcherrUrJrJrK_configure_launchersz#XLAFSDPStrategy._configure_launchercsN|jdusJt|jdkrtdt|jdd|_|jt_t dS)NrfzThe z does not support running on a single device with the PjRT runtime. Try using all devices or the `SingleDeviceXLAStrategy` strategyT) r0rTNotImplementedErrortype__name__rFrbr&rankr=setup_environmentrUrHrJrKrnsz!XLAFSDPStrategy.setup_environmentmodule optimizerscCstdt|jd)z]Returns NotImplementedError since for XLAFSDP optimizer setup must happen after module setup.zThe `z` does not support the joint setup of module and optimizer(s). Please do it in this order: Create the model, call `setup_module`, create the optimizer, call `setup_optimizer`.)rjrkrl)rGrorprJrJrKsetup_module_and_optimizerssz+XLAFSDPStrategy.setup_module_and_optimizerscsdddlm|}tfdd|Dr"d|vr"td|d=t|s0dd|i|}|S) NrXlaFullyShardedDataParallelc3s|]}t|VqdSrWrY).0modXLAFSDPrJrK sz/XLAFSDPStrategy.setup_module..r3zoA XLAFSDP `auto_wrap_policy` is set, but at least one submodule is already wrapped. The policy will be ignored.rorJ)torch_xla.distributed.fsdprs_parse_fsdp_kwargsanymodulesr'rY)rGror8rJrwrK setup_modules " zXLAFSDPStrategy.setup_modulecCsdSrWrJ)rGrorJrJrKmodule_to_devicesz XLAFSDPStrategy.module_to_device empty_initcCsD|j}|}t}|tt|d|||||S)N)enabled)r2module_init_contextmodule_sharded_contextr enter_contextr%bool)rGrprecision_init_ctxmodule_sharded_ctxstackrJrJrKrs   z#XLAFSDPStrategy.module_init_contextcCstSrW)rrUrJrJrKrsz&XLAFSDPStrategy.module_sharded_context dataloaderr,cCsDddlm}t||r |S|||j}|jj|_t|jdd|_|S)Nrr+ batch_sampler)%torch_xla.distributed.parallel_loaderr,rYrR_loaderdatasetgetattrr)rGrr,rJrJrKprocess_dataloaders    z"XLAFSDPStrategy.process_dataloader optimizercCs tdd|jDr |Std)aLSet up an optimizer for a model wrapped with XLAFSDP. 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. css*|]}|dD] }t|ddVqqdS)params _is_shardedFN)r)rugroupprJrJrKrys(z2XLAFSDPStrategy.setup_optimizer..zThe optimizer does not seem to reference any XLAFSDP parameters. HINT: Make sure to create the optimizer after setting up the model.)r| param_groups ValueError)rGrrJrJrKsetup_optimizers  zXLAFSDPStrategy.setup_optimizercKs.|jdi|}ddlmm}||S)a(Overrides default tpu optimizer_step since FSDP should not call `torch_xla.core.xla_model.optimizer_step`. Performs the actual optimizer step. Args: optimizer: the optimizer performing the step **kwargs: Any extra arguments to ``optimizer.step`` rNrJ)steprMrNrO mark_step)rGrr8lossrQrJrJrKoptimizer_steps zXLAFSDPStrategy.optimizer_step@Tmax_norm norm_typeerror_if_nonfinitecCs|j||j||dS)zClip gradients by norm.)rr)r2unscale_gradientsclip_grad_norm_)rGrorrrrrJrJrKclip_gradients_norms z#XLAFSDPStrategy.clip_gradients_normclip_valcCstd)zClip gradients by value.z~XLA's FSDP strategy does not support to clip gradients by value. Consider clipping by norm instead or choose another strategy!)rj)rGrorrrJrJrKclip_gradients_value,sz$XLAFSDPStrategy.clip_gradients_valuetensorr sync_gradscCs|js|St|tstdt|jd||dkr"|d}|j}| |j }ddl m m }ddlm m}|rD||n||}| |}|S)aCFunction to gather a tensor from several distributed processes. Args: tensor: tensor to all-gather. group: unused. sync_grads: flag that allows users to synchronize gradients for the all-gather operation. Return: A tensor of shape (world_size, ...) `z4.all_gather` is only implemented for tensors. Given rN)rFrYrrjrkrldim unsqueezedevicetorRtorch_xla.core.functionsrN functionsrMrO all_gather)rGrrroriginal_devicexfrQrJrJrKr4s      zXLAFSDPStrategy.all_gatheroutput reduce_opcCst|ts tj||jd}t|to|tjk}t|to!|dv}|s&|r-t d|ddl m m }| d|t}t|trM|dvrM||j}|S)N)r)summeanavgzaCurrently, the XLAFSDPStrategy only supports `sum`, `mean`, `avg` for the reduce operation, got: rreduce)rr)rYrtorchrrRr*SUMr<lowerrrMrNrO mesh_reducerrg)rGrrrinvalid_reduce_opinvalid_reduce_op_strrQrJrJrK all_reduceRs  zXLAFSDPStrategy.all_reducenameargscOs6|jsdSddlmm}|durd}||dS)Nr)rFrMrNrO rendezvous)rGrrr8rQrJrJrKbarrieris zXLAFSDPStrategy.barrierrobjsrccCs|js|Sddlmm}t|t}|r*|dkr |d}|j}| |j }nt }t ||t jt||j t jd}|g}|j||d|d}|sdt |}t |}|S| |}|S)Nr)rdtype) root_ordinal)rFrMrNrOrYrrrrrrRr\BytesIOrsaver bytearray getbufferfloatcollective_broadcastcpubytenumpyload)rGrrrQ is_tensorrbufferrJrJrK broadcastts.      zXLAFSDPStrategy.broadcastpathstatestorage_optionsfilterc st||}|rt|rtd|ddlmfdd|D}t |dkr3t dt |dkr=t ddd l m m }||j}|d usSJ|jrvtt |D]}||jkrk||||||d |d q\n||||||jd krt|d } d} t ||jkrtd| d| dddlm} |d|jr|jd} | | | t| |j|t| t| t||dd Sd S)aSave model, optimizer, and other state in the provided checkpoint directory. If the user specifies sharded checkpointing, the directory will contain one file per process, with model- and optimizer shards stored per file. If the user specifies full checkpointing, the directory will contain a consolidated checkpoint combining all of the sharded checkpoints. z:The checkpoint directory already exists and is not empty: rrrcsg|] }t|r|qSrJrt)rurorwrJrK sz3XLAFSDPStrategy.save_checkpoint..aCould not find a XLAFSDP 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.rfzFound multiple XLAFSDP modules 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.Nz wait-for-z-saver6 checkpointz_rank-*-of-*.pthaMultihost setups do not have a shared filesystem, so the checkpoint shards cannot be consolidated into a single checkpoint after saving them. Please switch to `XLAFSDPStrategy(state_dict_type='sharded')`. TIP: You can consolidate them manually by getting them together into a single directory and running `python -m torch_xla.distributed.fsdp.consolidate_sharded_ckpts --ckpt_prefix z --ckpt_suffix z* --save_path 'path/to/consolidated.ckpt'`.)%consolidate_sharded_model_checkpointsbefore_ckpt_consolidationzconsolidated.ckptafter_ckpt_consolidation) rris_dirr|iterdirFileExistsErrorrzrsvaluesrTrrMrNrOrr0rErangerd_save_checkpoint_shardrrDr<rgOSErrorris_global_zeroparentr1remove_checkpointr$mv) rGrrrrr}rQr0rm ckpt_prefix ckpt_suffixr save_pathrJrwrKsave_checkpointsX          zXLAFSDPStrategy.save_checkpointc Csddlm}i}|D].\}}t|tr%t||r%|} ||d<n t|tr/|} n|} t||p6i| |q |j j ||d|j dd|j dd|ddS) Nrrrshard_metadatacheckpoint_rank-08d-of-.pth)r) rzrsitemsrYr state_dictget_shard_metadatarrr1rrbrg) rGrrrrrxconverted_statekeyr convertedrJrJrKrs    z&XLAFSDPStrategy._save_checkpoint_shardstrictcs|s td|dt||}t|ttfrtdddlmfdd| D}dd| D}|j d kr|d |j d d |j d d }| sXtdt|dt|dkrbtdt|dkrltdt| d\}}t|} |j| d|d| D] \} } | | | q| ||} |||} t| | |d| D]}|| vr| |||<| |qi}t| r| D]}| |||<qd|vr|d|S|j dkrD| stdt|dt|dkst|||dkrtdt|dkrtdd|vs/t|d}tjjs3tdt|}|j|d|d|Std|j )aGiven a folder, load the contents from a checkpoint and restore the state of the given objects. The strategy currently only supports saving and loading sharded checkpoints which are stored in form of a directory of multiple files rather than a single file. z0Got `XLAFSDPStrategy.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, ...})`zmLoading a single module or optimizer object from a checkpoint is not supported yet with the XLAFSDP strategy.rrrcs i|] \}}t|r||qSrJrt)rurrorwrJrK  z3XLAFSDPStrategy.load_checkpoint..cSs i|] \}}t|tr||qSrJ)rYr)ruroptimrJrJrKrrr.rrrrz The path zv does not point to valid sharded checkpoints. Make sure the path points to a directory with XLAFSDP checkpoint shards.aCould not find a XLAFSDP 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.rfzFound multiple XLAFSDP modules 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.model)rrr6zt does not point to a valid full checkpoint. Make sure the path points to a directory with a full XLAFSDP checkpoint.ztLoading a full checkpoint will only load the full model. The optimizer and any additional metadata are not included.zmFound a XLAFSDP model in the provided checkpoint state. Please provide the model without any XLAFSDP wrapper.zFXLAFSDP only supports a single model instance with 'model' as the key.zUnknown state_dict_type: )rrrrYrrrjrzrsrrDrbrgis_filer<rTlistrrload_state_dictkeysr#removepopr'nn)rGrrrr}rpfile_ro sharded_ckptopt_keyoptloaded_metadata_keysrequested_metadata_keysrmetadatar full_ckptrJrwrKload_checkpoints           0" zXLAFSDPStrategy.load_checkpointstrategy_registrycCs|jd||jddS)Nxla_fsdp) description)registerrl)clsr rJrJrKregister_strategiesgsz#XLAFSDPStrategy.register_strategiescCs@|j}|j}t|tr|d|jt|j|}t |j |S)N compute_dtype) rCcopyr2rYr setdefault_desired_dtype_auto_wrap_policy_kwargsrA _activation_checkpointing_kwargsrB)rGr8r2rJrJrKr{ls    z"XLAFSDPStrategy._parse_fsdp_kwargs)NNNNNNr.F)r9NrW)rT)NF)NN)r)NT)Erl __module__ __qualname____doc__r rr rrrr_POLICY _POLICY_SETr rrr>propertyrrRintrVr1setterr2rbrdrergrirnrrrrqr~rr rrrrrr)rrrrrrrr*r<rrr rr(r rrrrr  classmethodrrr{ __classcell__rJrJrHrKr-5sB             $"  " " K "  gr-policyrr8r9cCs:|dur|St|trddlm}t||d}||d<|S)Nr)transformer_auto_wrap_policy)transformer_layer_clsr3)rYsettorch_xla.distributed.fsdp.wrapr$r)r#r8r$rJrJrKrxs   rrorcOsHddlm}ddlm}t|t|r||n|}||g|Ri|S)Nrrr)checkpoint_module)rzrsr(rYtuple)r#rorr8rxr(rJrJrK&_activation_checkpointing_auto_wrappers  r*cCsH|s|Sd|vr tdt|tstd|dtt|}||d<|S)Nauto_wrapper_callablez]You cannot set both `auto_wrapper_callable` and `activation_checkpointing_policy`. Choose onez7`activation_checkpointing_policy` must be a set, found zC. You can try defining and passing `auto_wrapper_callable` instead.)rrYr&r^rr*)r#r8r+rJrJrKrs   rc@s&eZdZedededefddZdS)r?rorr9cCsF|stSddlm}t||std|jjd|jjd|S)z|Blocks gradient synchronization inside the :class:`~torch_xla.distributed.fsdp.XlaFullyShardedDataParallel` wrapper.rrrzABlocking backward sync is only possible if the module passed to `zE.no_backward_sync` is wrapped in `XlaFullyShardedDataParallel`. Got: .)rrzrsrYr^rIrlno_sync)rGrorrxrJrJrKno_backward_syncs  z,_XLAFSDPBackwardSyncControl.no_backward_syncN)rlrrrrrr r.rJrJrJrKr?sr?)Kr\ contextlibrr functoolsrpathlibrtypingrrrr r r r r rrrrrrtorch.nnr torch.optimrtorch.utils.datartyping_extensionsrlightning.fabric.acceleratorsr!lightning.fabric.accelerators.xlarlightning.fabric.pluginsr%lightning.fabric.plugins.environmentsrlightning.fabric.plugins.io.xlarlightning.fabric.strategiesrr lightning.fabric.strategies.fsdpr)lightning.fabric.strategies.launchers.xlar$lightning.fabric.strategies.strategyr r!r"r##lightning.fabric.utilities.cloud_ior$lightning.fabric.utilities.initr%$lightning.fabric.utilities.rank_zeror&r' lightning.fabric.utilities.typesr(r)r*rr,rrrrr-rr*rr?rJrJrJrKsJ   8                G