a
    dx#                     @   s0  d dl Z d dlZd dlmZmZmZmZmZmZm	Z	m
Z
 d dlZd dlmZ d dlmZ d dlmZ d dlmZ d dlmZ d dlmZ d d	lmZ d d
lmZ d dlmZ d dlmZ d dlm Z  d dl!m"Z" d dl#m$Z$ d dl%m&Z& d dl'm(Z( d dl)m*Z*m+Z+ e	rerd dl,m-Z- G dd deZ.dS )    N)AnyDictListMappingOptionalSequenceTYPE_CHECKINGUnion)Tensor)Module)
DataLoader)Accelerator)_XLA_AVAILABLE)XLAEnvironment)CheckpointIO)XLACheckpointIO)	Precision)ParallelStrategy)_XLALauncher)
TBroadcast)apply_to_collection)has_len)rank_zero_only)_PATHReduceOpMpDeviceLoaderc                       s  e Zd ZdZd:ee eeej  ee	 ee
 dd fddZeejdddZeedd	d
Zee	dddZejee	 ddddZeedddZddddZdd fddZeedddZeddddZeddddZd;eee eedd d!Zd<eeef ee eeee f  ed"d#d$Z!d=ee  eedd%d&d'Z"d>e#ee#d)d*d+Z$d?e%e ef e&ee dd,d-d.Z'e&dd/d0d1Z(e)e%dd2d3d4Z*ddd5d6Z+e,edd7d8d9Z-  Z.S )@XLAStrategyzxStrategy for training multiple TPU devices using the :func:`torch_xla.distributed.xla_multiprocessing.spawn`
    method.N)acceleratorparallel_devicescheckpoint_io	precisionreturnc                    s,   t  j||t ||d |  d | _d| _d S )N)r   r   cluster_environmentr    r!   F)super__init__r   Z_backward_sync_control	_launched)selfr   r   r    r!   	__class__ h/var/www/html/stable-diffusion-webui/venv/lib/python3.9/site-packages/lightning_fabric/strategies/xla.pyr%   -   s    zXLAStrategy.__init__)r"   c                 C   s(   | j stddd lm  m} | S )NzFAccessing the XLA device before processes have spawned is not allowed.r   )r&   RuntimeErrortorch_xla.core.xla_modelcore	xla_modelZ
xla_device)r'   xmr*   r*   r+   root_device?   s    zXLAStrategy.root_devicec                 C   s   | j d urt| j S dS Nr   )r   lenr'   r*   r*   r+   num_processesG   s    zXLAStrategy.num_processesc                 C   s   | j d u rt | _ | j S N)_checkpoint_ior   r4   r*   r*   r+   r    K   s    
zXLAStrategy.checkpoint_io)ior"   c                 C   s
   || _ d S r6   )r7   )r'   r8   r*   r*   r+   r    Q   s    c                 C   s(   dd l m  m} |jtjv o&| jdkS )Nr      )Ztorch_xla.core.xla_env_varsr.   Zxla_env_varsZHOST_WORLD_SIZEosenviron
world_size)r'   Zxenvr*   r*   r+   _is_distributedU   s    zXLAStrategy._is_distributedc                 C   s   t | | _d S r6   )r   Z	_launcherr4   r*   r*   r+   _configure_launcher\   s    zXLAStrategy._configure_launcherc                    s$   d| _ |   | jt_t   d S )NT)r&   _set_world_ranksglobal_rankr   rankr$   setup_environmentr4   r(   r*   r+   rB   _   s    zXLAStrategy.setup_environment)moduler"   c                 C   s   |S r6   r*   r'   rC   r*   r*   r+   setup_modulee   s    zXLAStrategy.setup_modulec                 C   s   | | j d S r6   )tor1   rD   r*   r*   r+   module_to_deviceh   s    zXLAStrategy.module_to_devicer   
dataloaderr"   c                 C   sN   t | ddlm} t||r$|S ||| j}|jj|_t|jdd |_	|S )Nr   r   batch_sampler)
r   _validate_dataloader%torch_xla.distributed.parallel_loaderr   
isinstancer1   Z_loaderZdatasetgetattrrJ   )r'   rI   r   r*   r*   r+   process_dataloaderk   s    


zXLAStrategy.process_dataloaderF)tensorgroup
sync_gradsr"   c                 C   s\   t |tr | dkr |d}ddlm  m} ddlm  m} |rR|	|S |	|S )a_  Function to gather a tensor from several distributed processes.

        Args:
            tensor: tensor of shape (batch, ...)
            group: not available with TPUs
            sync_grads: flag that allows users to synchronize gradients for the all_gather operation
        Return:
            A tensor of shape (world_size, batch, ...)
        r   N)
rM   r
   ZdimZ	unsqueezeZtorch_xla.core.functionsr.   Z	functionsr-   r/   
all_gather)r'   rP   rQ   rR   Zxfr0   r*   r*   r+   rS   y   s
    

zXLAStrategy.all_gather)outputrQ   	reduce_opr"   c                 C   s   t |tstj|| jd}t |to,|tjk}t |toB| dv}|sL|rZt	d| dd l
m  m} |d|t}t |tr| dv r|| j }|S )N)device)summeanavgz]Currently, the XLAStrategy only supports `sum`, `mean`, `avg` for the reduce operation, got: r   reduce)rY   rX   )rM   r
   torchrP   r1   r   ZSUMstrlower
ValueErrorr-   r.   r/   Zmesh_reducerW   r<   )r'   rT   rQ   rU   Zinvalid_reduce_opZinvalid_reduce_op_strr0   r*   r*   r+   
all_reduce   s    

zXLAStrategy.all_reduce)nameargskwargsr"   c                 O   s&   | j r"dd lm  m} || d S r2   )r=   r-   r.   r/   Z
rendezvous)r'   r`   ra   rb   r0   r*   r*   r+   barrier   s    zXLAStrategy.barrierr   )objsrcr"   c                 C   s~   | j s
|S t }t|| t| }tj|| jtj	d}dd l
m  m} ||}t|   }t|}|S )N)rV   dtyper   )r=   r8   BytesIOr[   save	bytearray	getbufferrP   r1   floatr-   r.   r/   rS   cpubytenumpyload)r'   rd   re   bufferdataZdata_tensorr0   r*   r*   r+   	broadcast   s    

zXLAStrategy.broadcast)
checkpointfilepathstorage_optionsr"   c                 C   s   | j j|||d dS )a>  Save model/training states as a checkpoint file through state-dump and file-write.

        Args:
            checkpoint: dict containing model and trainer state
            filepath: write-target file's path
            storage_options: parameter for how to save to storage, passed to ``CheckpointIO`` plugin
        )ru   N)r    save_checkpoint)r'   rs   rt   ru   r*   r*   r+   rv      s    zXLAStrategy.save_checkpoint)rt   r"   c                 C   s   | j dkr| j| dS )zpRemove checkpoint filepath from the filesystem.

        Args:
            filepath: Path to checkpoint
        r   N)Z
local_rankr    remove_checkpoint)r'   rt   r*   r*   r+   rw      s    
zXLAStrategy.remove_checkpoint)strategy_registryr"   c                 C   s,   |j d| | jjd |j d| | jjd d S )NZ	tpu_spawn)descriptionZxla)registerr)   __name__)clsrx   r*   r*   r+   register_strategies   s    zXLAStrategy.register_strategiesc                 C   s   | j d u rd S | j  t_d S r6   )r#   r@   r   rA   r4   r*   r*   r+   r?      s    
zXLAStrategy._set_world_ranks)dataloadersr"   c                 C   s(   t d ddd}t| tttf|d d S )NrH   c                 S   s   t | stdd S )NzTPUs do not currently support IterableDataset objects, the dataset must implement `__len__`. HINT: You can mock the length on your dataset to bypass this MisconfigurationException.)r   	TypeError)rI   r*   r*   r+   check_has_len   s    z7XLAStrategy._validate_dataloader.<locals>.check_has_len)rf   Zwrong_dtypefunction)r   r   objectr   r   )r~   r   r*   r*   r+   rK      s    z XLAStrategy._validate_dataloader)NNNN)NF)NN)N)r   )N)/r{   
__module____qualname____doc__r   r   r   r[   rV   r   r   r%   propertyr1   intr5   r    setterboolr=   r>   rB   r   rE   rG   r   rO   r
   r   rS   r	   r   r\   r_   rc   r   rr   r   r   rv   rw   classmethodr}   r?   staticmethodrK   __classcell__r*   r*   r(   r+   r   )   s\         	r   )/r8   r:   typingr   r   r   r   r   r   r   r	   r[   r
   Ztorch.nnr   Ztorch.utils.datar   Zlightning_fabric.acceleratorsr   Z!lightning_fabric.accelerators.tpur   Z%lightning_fabric.plugins.environmentsr   Z)lightning_fabric.plugins.io.checkpoint_ior   Zlightning_fabric.plugins.io.xlar   Z"lightning_fabric.plugins.precisionr   Zlightning_fabric.strategiesr   Z)lightning_fabric.strategies.launchers.xlar   Z$lightning_fabric.strategies.strategyr   Z%lightning_fabric.utilities.apply_funcr   Zlightning_fabric.utilities.datar   Z$lightning_fabric.utilities.rank_zeror   Z lightning_fabric.utilities.typesr   r   rL   r   r   r*   r*   r*   r+   <module>   s,   (