o zi,@sddlZddlmZmZmZmZmZmZmZddl Z ddl m Z ddl m Z ddl mZddlmZddlmZddlmZdd lmZdd lmZdd lmZdd lmZdd lmZmZddl m!Z!ddl"m#Z#ddl$m%Z%ddl&m'Z'm(Z(erddl)m*Z*GdddeZ+dS)N) TYPE_CHECKINGAnyCallableDictListOptionalUnion)Tensor)Module) Optimizer) DataLoader)override) Accelerator)_XLA_GREATER_EQUAL_2_1) XLAPrecision)XLAEnvironment)XLACheckpointIO)ParallelStrategy_StrategyRegistry) _XLALauncher) TBroadcast)rank_zero_only)_PATHReduceOpMpDeviceLoadercseZdZdZ     dIdeedeeejdee dee de d df fd d Z e ed ejfd d Ze d efddZe ed e fddZejedee d dfddZe ed e fddZejedee d dfddZe ed effdd Ze ed effdd Ze ed effdd Ze ed effdd ZedJdd ZedJfd!d" Zed#ed efd$d%Zed#ed dfd&d'Zed(ed d)fd*d+ZedKd-e d.ee!d/e d e fd0d1Z"e dLd2e#e e!fd.ee!d3ee#e$e%fd e fd4d5Z&edMd6ee%d7e!d8e!d dfd9d:Z'edNdd?Z)e  dLd@e*dAe+e%e#ee,e!ffdBee!dCee+e%e-e%e!ge ffd df fdDdE Z.e/edFe0d dfdGdHZ1Z2S)O XLAStrategyzxStrategy for training multiple TPU devices using the :func:`torch_xla.distributed.xla_multiprocessing.spawn` method.NT acceleratorparallel_devices checkpoint_io precisionsync_module_statesreturncs.tj||t||dd|_d|_||_dS)N)rrcluster_environmentrr F)super__init__r_backward_sync_control _launched_sync_module_states)selfrrrr r! __class__S/home/ubuntu/.local/lib/python3.10/site-packages/lightning_fabric/strategies/xla.pyr%+s zXLAStrategy.__init__cCs(|jstdddlmm}|S)NzFAccessing the XLA device before processes have spawned is not allowed.r)r' RuntimeErrortorch_xla.core.xla_modelcore xla_model xla_device)r)xmr,r,r- root_device>szXLAStrategy.root_devicecCs|jdur t|jSdSNr)rlenr)r,r,r- num_processesGszXLAStrategy.num_processescCs&|j}|durt|tsJ|StSN)_checkpoint_io isinstancerr)pluginr,r,r-rKs zXLAStrategy.checkpoint_ioiocC*|durt|tstd|||_dS)NzHThe XLA strategy can only work with the `XLACheckpointIO` plugin, found )r;r TypeErrorr:)r)r>r,r,r-rT cCs(|j}|durt|tsJ|StdS)Nz32-true) _precisionr;rr<r,r,r-r [s zXLAStrategy.precisioncCr?)NzEThe XLA strategy can only work with the `XLAPrecision` plugin, found )r;rr@rB)r)r r,r,r-r drAc|jrtjSdSr5)r'r$ global_rankr7r*r,r-rDkzXLAStrategy.global_rankcrCr5)r'r$ local_rankr7r*r,r-rFprEzXLAStrategy.local_rankcrCr5)r'r$ node_rankr7r*r,r-rGurEzXLAStrategy.node_rankcrC)N)r'r$ world_sizer7r*r,r-rIzrEzXLAStrategy.world_sizecCst||_dSr9)r _launcherr7r,r,r-_configure_launcherszXLAStrategy._configure_launchercsN|jdusJt|jdkrtdt|jdd|_|jt_t dS)NrHzThe z does not support running on a single device with the PjRT runtime. Try using all devices or the `SingleDeviceXLAStrategy` strategyT) rr6NotImplementedErrortype__name__r'rDrrankr$setup_environmentr7r*r,r-rPszXLAStrategy.setup_environmentmodulecCs0|jrtr ddlm}nddlm}|||S)Nr)broadcast_master_param)r(rr/rRtorch_xla.experimental.pjrt)r)rQrRr,r,r- setup_modules  zXLAStrategy.setup_modulecCs||jdSr9)tor4)r)rQr,r,r-module_to_deviceszXLAStrategy.module_to_device dataloaderrcCsDddlm}t||r |S|||j}|jj|_t|jdd|_|S)Nrr batch_sampler)%torch_xla.distributed.parallel_loaderrr;r4_loaderdatasetgetattrrX)r)rWrr,r,r-process_dataloaders    zXLAStrategy.process_dataloaderFtensorgroup 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)r'r;r rLrMrNdim unsqueezedevicerUr4torch_xla.core.functionsr0 functionsr/r1 all_gather)r)r^r_r`original_devicexfr3r,r,r-rgs      zXLAStrategy.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)rd)summeanavgz]Currently, the XLAStrategy only supports `sum`, `mean`, `avg` for the reduce operation, got: rreduce)rnrm)r;r torchr^r4rSUMstrlower ValueErrorr/r0r1 mesh_reducerlrI)r)rjr_rkinvalid_reduce_opinvalid_reduce_op_strr3r,r,r- all_reduces  zXLAStrategy.all_reducenameargskwargscOs6|jsdSddlmm}|durd}||dS)Nr)r'r/r0r1 rendezvous)r)ryrzr{r3r,r,r-barriers zXLAStrategy.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)rddtype) root_ordinal)r'r/r0r1r;r rbrcrdrUr4r>BytesIOrpsaver^ bytearray getbufferfloatcollective_broadcastcpubytenumpyload)r)rrr3 is_tensorrhbufferr,r,r- broadcasts.      zXLAStrategy.broadcastpathstatestorage_optionsfiltercs2ddlmm}|tj||||ddS)aSave model, optimizer, and other state as a checkpoint file. Args: path: A path to where the file(s) should be saved state: A dictionary with contents to be saved. If the dict contains modules or optimizers, their state-dict will be retrieved and converted automatically. storage_options: Additional options for the ``CheckpointIO`` plugin filter: An optional dictionary of the same format as ``state`` mapping keys to callables that return a boolean indicating whether the given parameter should be saved (``True``) or filtered out (``False``). rN)rr)r/r0r1 mark_stepr$save_checkpoint)r)rrrrr3r*r,r-rszXLAStrategy.save_checkpointstrategy_registrycCs|jd||jddS)Nxla) description)registerrN)clsrr,r,r-register_strategies-szXLAStrategy.register_strategies)NNNNT)r"N)NF)NNr9)r)3rN __module__ __qualname____doc__rrrrprdrrboolr%propertyr r4intr8rsetterr rDrFrGrIrKrPr rTrVr r]r rrgrrrrrxr~rrrrr rr classmethodrr __classcell__r,r,r*r-r's    "  " "r),r>typingrrrrrrrrpr torch.nnr torch.optimr torch.utils.datar typing_extensionsr lightning_fabric.acceleratorsr!lightning_fabric.accelerators.xlarlightning_fabric.pluginsr%lightning_fabric.plugins.environmentsrlightning_fabric.plugins.io.xlarlightning_fabric.strategiesrr)lightning_fabric.strategies.launchers.xlar$lightning_fabric.strategies.strategyr$lightning_fabric.utilities.rank_zeror lightning_fabric.utilities.typesrrrYrrr,r,r,r-s* $