iu $ddlZddlmZmZmZddlmZddlmZddl m Z m Z m Z m Z mZmZddlZddlmZddlmZddlmZdd lmZdd lmZdd lmZdd lmZdd lmZm Z m!Z!ddl"m#Z#ddl$m%Z%ddl&m'Z'm(Z(ddl)m*Z*ddl+m,Z,ddl-m.Z.m/Z/m0Z0m1Z1ddl2m3Z3ddl4m5Z5ddl6m7Z7m8Z8ddl9m:Z:m;Z;mZ>ddl?m@Z@eAeBeZCeeCe eeDeEgeDffZFGdde'e0ZGdeddeHdeHfd ZIdeCd!ed"e de def d#ZJdeeCdeHdeHfd$ZKGd%d&e/ZLy)'N)AbstractContextManager ExitStack nullcontext)partial)Path) TYPE_CHECKINGAnyCallableLiteralOptionalUnion)Tensor)Module) Optimizer) DataLoader)override) Accelerator)_XLA_AVAILABLE) CheckpointIO Precision 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 OptimizableReduceOp) _LRSchedulerMpDeviceLoaderceZdZdZ dKdeedeeejdee dee dee dee d e d d ed ed dffd Zeed ejfdZed efdZeed e fdZej.edeed dfdZeed e fdZej.edeed dfdZeed effd Zeed effd Zeed effd Zeed effd ZedLdZedLfd Z e dMde!dee"dedd e#e!ee"edffd Z$ede!d e!fd!Z%ede!d dfd"Z&dMd#eed e'fd$Z(ed e'fd%Z)ed&e*d d'fd(Z+ed)e"d e"fd*Z,ed)e-d ed efd+Z.e dNde!d)e"d,e/e0efd-e/e0efd.ed e1f d/Z2ede!d)e"d0e/e0efd dfd1Z3edOd2e1d3eed4ed e1fd5Z4e dPd6e/e1efd3eed7ee/e5e6fd e1fd8Z7edMd9ee6d:ed ed dfd;Z8edQdZ:e dPd?e;d@ed?e?d@e` feature. For more information check out https://github.com/pytorch/xla/blob/v2.5.0/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`. N acceleratorparallel_devices checkpoint_io precisionauto_wrap_policyactivation_checkpointing_policystate_dict_type)fullshardedsequential_savekwargsreturnc  tstttt |||t ||t |_||_||_ | |_ ||_ ||_ d|_ y)N)r.r/cluster_environmentr0r1F)rModuleNotFoundErrorstrsuper__init__r_XLAFSDPBackwardSyncControl_backward_sync_control_auto_wrap_policy _activation_checkpointing_policy _fsdp_kwargs_state_dict_type_sequential_save _launched) selfr.r/r0r1r2r3r4r7r8 __class__s y/Volumes/fast/ai/experiments/voice-extract-mac/.venv/lib/python3.12/site-packages/lightning_fabric/strategies/xla_fsdp.pyr?zXLAFSDPStrategy.__init__Tsy%c.&9: : #- . 0'  'B&C#!10O-" / /cb|js tdddlmcm}|j S)NzFAccessing the XLA device before processes have spawned is not allowed.r)rG RuntimeErrortorch_xla.core.xla_modelcore xla_model xla_device)rHxms rJ root_devicezXLAFSDPStrategy.root_devicers)~~gh h--}}rKcH|jt|jSdSNr)r/lenrHs rJ num_processeszXLAFSDPStrategy.num_processes{s$-1-B-B-Ns4(()UTUUrKcZ|j}|t|tsJ|StSN)_checkpoint_io isinstancerrHplugins rJr0zXLAFSDPStrategy.checkpoint_ios3$$  fo6 66M  rKiocR|t|tstd|||_y)NzHThe XLA strategy can only work with the `XLACheckpointIO` plugin, found )r\r TypeErrorr[)rHr_s rJr0zXLAFSDPStrategy.checkpoint_ios/ >*R"Afgifjkl l rKc\|j}|t|tsJ|StdS)Nz32-true) _precisionr\rr]s rJr1zXLAFSDPStrategy.precisions3  fl3 33MI&&rKcR|t|tstd|||_y)NzJThe XLA FSDP strategy can only work with the `XLAPrecision` plugin, found )r\rrarc)rHr1s rJr1zXLAFSDPStrategy.precisions/  I|)Lhirhstu u#rKc4|jr t|SdSrU)rGr> global_rankrHrIs rJrfzXLAFSDPStrategy.global_ranks'+nnuw";!;rKc4|jr t|SdSrU)rGr> local_rankrgs rJrizXLAFSDPStrategy.local_rank&*^^uw!::rKc4|jr t|SdSrU)rGr> node_rankrgs rJrlzXLAFSDPStrategy.node_ranks%)NNuw 99rKc4|jr t|SdS)N)rGr> world_sizergs rJrozXLAFSDPStrategy.world_sizerjrKc$t||_yrZ)r _launcherrWs rJ_configure_launcherz#XLAFSDPStrategy._configure_launchers%d+rKc|jJt|jdk(r"tdt|jdd|_|j t_t|)y)NrnzThe z does not support running on a single device with the PjRT runtime. Try using all devices or the `SingleDeviceXLAStrategy` strategyT) r/rVNotImplementedErrortype__name__rGrfr$rankr>setup_environmentrgs rJrxz!XLAFSDPStrategy.setup_environmentsw$$000 t$$ % *&tDz**+,SS  ".. !#rKmodule optimizers schedulerr)cFtdt|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`.)rtrurv)rHryrzr{s rJsetup_module_and_optimizersz+XLAFSDPStrategy.setup_module_and_optimizerss- "DJ''()' '  rKcddlm|j}tfd|j Drd|vrt d|d=t |s dd|i|}|S)NrXlaFullyShardedDataParallelc36K|]}t|ywrZ)r\).0modXLAFSDPs rJ z/XLAFSDPStrategy.setup_module..sD3CCz#w'3Csr2zoA XLAFSDP `auto_wrap_policy` is set, but at least one submodule is already wrapped. The policy will be ignored.ry)torch_xla.distributed.fsdpr_parse_fsdp_kwargsanymodulesr%r\)rHryr8rs @rJ setup_modulezXLAFSDPStrategy.setup_moduleskU((* D6>>3CD DI[_eIe / )*&'*5F5f5F rKcyrZr)rHrys rJmodule_to_devicez XLAFSDPStrategy.module_to_devices rK empty_initc|jj}|j}t}|j t t ||j ||j ||S)N)enabled)r1module_init_contextmodule_sharded_contextr enter_contextr#bool)rHrprecision_init_ctxmodule_sharded_ctxstacks rJrz#XLAFSDPStrategy.module_init_contextsf!^^??A!88:  JtJ/?@A ./ ./ rKctSrZ)rrWs rJrz&XLAFSDPStrategy.module_sharded_contexts }rK dataloaderr+cddlm}t||r|S|||j}|jj |_t |jdd|_|S)Nrr* batch_sampler)%torch_xla.distributed.parallel_loaderr+r\rS_loaderdatasetgetattrr)rHrr+s rJprocess_dataloaderz"XLAFSDPStrategy.process_dataloadersYH j. 1 #J0@0@A '//77 #*:+=+=PT#U  rK optimizercTtd|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. c3JK|]}|dD]}t|ddyw)params _is_shardedFN)r)rgroupps rJrz2XLAFSDPStrategy.setup_optimizer.. s.m=SE]bck]lXYwq-/]l/=Ss!#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)rHrs rJsetup_optimizerzXLAFSDPStrategy.setup_optimizers0 mY=S=Sm m  +  rKc \|jdi|}ddlmcm}|j |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`` rNr)steprNrOrP mark_step)rHrr8lossrRs rJoptimizer_stepzXLAFSDPStrategy.optimizer_steps+y~~''--  rKmax_norm norm_typeerror_if_nonfinitec|jj|t|jsJ|j||S)zClip gradients by norm.)rr)r1unscale_gradientscallableclip_grad_norm_)rHryrrrrs rJclip_gradients_normz#XLAFSDPStrategy.clip_gradients_norm s@ ((3..///%%x9%MMrKclip_valctd)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!)rt)rHryrrs rJclip_gradients_valuez$XLAFSDPStrategy.clip_gradients_value.s" M  rKtensorr sync_gradsc|js|St|ts$tdt |j d||j dk(r|jd}|j}|j|j}ddl m cm }ddlm cm}|r|j!|n|j!|}|j|}|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)rGr\rrtrurvdim unsqueezedevicetorStorch_xla.core.functionsrO functionsrNrP all_gather)rHrrroriginal_devicexfrRs rJrzXLAFSDPStrategy.all_gather6s~~M&&)%DJ''((\]c\de  ::<1 %%a(F --4++,----*4v&"--:O?+ rKoutput reduce_opct|ts!tj||j}t|t xr|t j k7}t|txr|jdv}|s|rtd|ddl m cm }|jd|t}t|tr!|jdvr||jz }|S)N)r)summeanavgzaCurrently, the XLAFSDPStrategy only supports `sum`, `mean`, `avg` for the reduce operation, got: rreduce)rr)r\rtorchrrSr(SUMr=lowerrrNrOrP mesh_reducerro)rHrrrinvalid_reduce_opinvalid_reduce_op_strrRs rJ all_reducezXLAFSDPStrategy.all_reduceTs&&)\\&1A1ABF&y(;Y X\\@Y *9c : ny?PXn?n  5;   .-&#6 i %)//*;*Ndoo-F rKnameargscZ|jsyddlmcm}|d}|j |y)Nr)rGrNrOrP rendezvous)rHrrr8rRs rJbarrierzXLAFSDPStrategy.barrierks(~~ -- <D drKobjsrcc|js|Sddlmcm}t |t }|rL|j dk(r|jd}|j}|j|j}nqtj}tj||tjt!|j#|jtj$}|g}|j'|||d}|sVtj|j)j+j-}tj.|}|S|j}|S)Nr)rdtype) root_ordinal)rGrNrOrPr\rrrrrrSr_BytesIOrsaver bytearray getbufferfloatcollective_broadcastcpubytenumpyload)rHrrrR is_tensorrbuffers rJ broadcastzXLAFSDPStrategy.broadcastvs~~J--sF+ wwyA~mmA&!jjO&&))*CZZ\F JJsF #,,&**,-d6F6FekkCe #6!fZZ  0 6 6 89F**V$C &&)C rKpathstatestorage_optionsfilterczt|j|}|jr't|j rt d|ddlm}|jDcgc]}t||s|}}t|dk(r tdt|dkDr tdddl m cm}|j|j } | J|j"rRt%t| D]:} | |j&k(r|j)|||||j+d| d <n|j)|||||j,d k(rt/|d z } d } t| |j0k7rt3d | d| dddlm} |j+d|j6rj|j8dz }| | | t/||j:j=|t?|jAt/|t/||j+dyycc}w)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: rraCould 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.rnzFound 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-saver5 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_dirriterdirFileExistsErrorrrvaluesr\rVrrNrOrPrr/rFrangeri_save_checkpoint_shardrrEr=roOSErrorris_global_zeroparentr0remove_checkpointr"mv)rHrrrrrryrrRr/rw ckpt_prefix ckpt_suffixr save_paths rJsave_checkpointzXLAFSDPStrategy.save_checkpoints( DNN4() ;;=S0!$^_c^d"ef fU(- Vf*VW:U6V w<1 s  wt|j)|j)z | j)z dkDr t1dt|dkDr tdd|vs)t|dx}t"j2js t dt#j$||}|j'|j/d||Std|jcc}}wcc} }w)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.rrr6rr r r z 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.rnzFound 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)rrr5zt 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.)rzUnknown state_dict_type: )rrrr\rrrtrrr rErfrois_filer=rVlistrrload_state_dictkeysr!removepopr%nn)rHrrrrrrryroptimrzfile_ sharded_ckptopt_keyoptloaded_metadata_keysrequested_metadata_keysmetadatar full_ckpts rJload_checkpointzXLAFSDPStrategy.load_checkpointsB5)LTT DNN4() efi0 1%C  V27++-_-;3:fV]C^3;-_38;;=a=ZS%JuV_D`c5j= a  I -,T-=-=c,B$tWZF[[_``D<<> D }-CC7|q  x 7|a T W]]_-a0IAv ::d+L  " "<#8 " H * 0 0 2 ##L$9:!3$0#4#4#6#G*//J[#[ &+jjlW\\^&CjooFW&W # -.EG[dj k...!-c!2E#J(//4/ H'(/C$0$5HSM0 8+ -.O  F *<<> D }-AA:"c%**,*G*//J[*[&\_`&`S7|a Me#:uW~6Meuxx+_)*rss 4lCI  ! !)--"8 ! H 4T5J5J4KLMMO`as&N6;N6N</N<strategy_registryc@|jd||jy)Nxla_fsdp) description)registerrv)clsr(s rJregister_strategiesz#XLAFSDPStrategy.register_strategiesjs "":s "MrKc|jj}|j}t|tr|j d|j t|j|}t|j|S)N compute_dtype) rDcopyr1r\r setdefault_desired_dtype_auto_wrap_policy_kwargsrB _activation_checkpointing_kwargsrC)rHr8r1s rJrz"XLAFSDPStrategy._parse_fsdp_kwargsosg""'')NN i .   oy/G/G H)$*@*@&I/0U0UW]^^rK)NNNNNNr6F)r9NrZ)g@T)NF)NN)r)NTN)Grv __module__ __qualname____doc__r rrrrrr_POLICY _POLICY_SETr rr r?propertyrrSintrXr0setterrr1rrfrirlrorrrxrrtupler}rrrrrrrrr'rr rrrrrr(r=rrrrr&dictr rrrr' classmethodrr.r __classcell__)rIs@rJr-r-6s>.29=37,0.2AE6? %k*#4 #56 0  L)  #7+ *2+)>!!23 < U\\VsVV !!! ! 6!4!!  '<'' $8I#6$4$$   Z ? I5  I5I5V  Cvy#5667 "#  c8S#J,<#==>?    8_c'+ fNfNfic5TWAW;X6X1YYZ[fN fN tn fN c3h fNfNP N4EN$NN _D _rKr-policyr9r8r9c^||St|trddlm}t ||}||d<|S)Nr)transformer_auto_wrap_policy)transformer_layer_clsr2)r\settorch_xla.distributed.fsdp.wraprDr)rBr8rDs rJr4r4{s8 ~ &#P5VT!'F  MrKryrcpddlm}ddlm}t|t |r||n|}||g|i|S)Nrr)checkpoint_module)rrrIr\r>)rBryrr8rrIs rJ&_activation_checkpointing_auto_wrapperrJs8Q<*4VU6]*K v &QWF 6 +D +F ++rKc|s|Sd|vr tdt|tstd|dt t |}||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.)rr\rFrarrJ)rBr8rLs rJr5r5sm  &( k   fc "EfXN8 8  $$JFS&;F "# MrKc*eZdZedededefdZy)r@ryrr9c|s tSddlm}t||s:t d|j j d|j j d|jS)z|Blocks gradient synchronization inside the :class:`~torch_xla.distributed.fsdp.XlaFullyShardedDataParallel` wrapper.rrzABlocking backward sync is only possible if the module passed to `zE.no_backward_sync` is wrapped in `XlaFullyShardedDataParallel`. Got: .)rrrr\rarIrvno_sync)rHryrrs rJno_backward_syncz,_XLAFSDPBackwardSyncControl.no_backward_syncsk= U&'*^^,,-.))22316  ~~rKN)rvr6r7rrrrrQrrKrJr@r@s*   v    AW    rKr@)Mr_ contextlibrrr functoolsrpathlibrtypingrr r r r r rrtorch.nnr torch.optimrtorch.utils.datartyping_extensionsrlightning_fabric.acceleratorsr!lightning_fabric.accelerators.xlarlightning_fabric.pluginsrrr%lightning_fabric.plugins.environmentsrlightning_fabric.plugins.io.xlarlightning_fabric.strategiesrr lightning_fabric.strategies.fsdpr)lightning_fabric.strategies.launchers.xlar$lightning_fabric.strategies.strategyrrr r!#lightning_fabric.utilities.cloud_ior"lightning_fabric.utilities.initr#$lightning_fabric.utilities.rank_zeror$r% lightning_fabric.utilities.typesr&r'r(torch.optim.lr_schedulerr)rr+rFrur:rr<r9r-r?r4rJr5r@rrKrJrhs& EEII !'&5<JJ@;K:B ?6OII5D$v,  XvtS&94&?@@ AB _&B _J Xi%8 $ 4 ,;,,WZ,fi,nt,Xk-BDUY" "6 rK