a
    d65                     @   s  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Zd dlmZ d dlmZmZ d d	lmZ d d
lmZ d dlmZ d dl m!Z!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/m0Z0 d dl1m2Z2 d dl3m4Z4m5Z5 d dl6m7Z7 d dl8m9Z9 d dl:m;Z;m<Z<m=Z= e	r~er~d dl>m?Z? ndZ?G dd de*Z@dS )    N)AnyDictListMappingOptionalSequenceTYPE_CHECKINGUnion)apply_to_collection)Tensor)Module)
DataLoader)_XLA_AVAILABLE)CheckpointIOXLACheckpointIO)XLAEnvironment)has_len)_optimizers_to_device)_PATHReduceOp)LightningDistributedModule)_WrappingCheckpointIO)PrecisionPlugin)DDPSpawnStrategy)_XLALauncher)
TBroadcast)DataConnector)	TrainerFn)find_shared_parametersset_shared_parameters)MisconfigurationException)rank_zero_only)EVAL_DATALOADERSSTEP_OUTPUTTRAIN_DATALOADERSMpDeviceLoaderc                	       s  e Zd ZdZdZd[ed eeej  ee	 ee
 eedd fddZee	d	d
dZejee	 ddddZeejd	ddZeed	ddZeeeef ddddZedddddZddd fddZdd	ddZddddd Zeedd!d"Zeee ef d	d#d$Z!eed	d%d&Z"e#d'd(d)d*Z$dd	d+d,Z%dd	d-d.Z&d\ee  eedd/d0d1Z'd]e(ee(d3d4d5Z)d^ee*ef ee eee+e f  e*d6d7d8Z,dd	d9d:Z-dd	d;d<Z.eeee/ d=d>d?Z0eeee/ d=d@dAZ1eee/d=dBdCZ2e/e/dDdEdFZ3e/e/dDdGdHZ4e/e/dDdIdJZ5dd	dKdLZ6d_ee ef e7ee ddMdNdOZ8e7ddPdQdRZ9d`e*ee ee*dSdTdUZ:dd	 fdVdWZ;e<eddXdYdZZ=  Z>S )aTPUSpawnStrategyzxStrategy for training multiple TPU devices using the :func:`torch_xla.distributed.xla_multiprocessing.spawn`
    method.Z	tpu_spawnNFzpl.accelerators.Accelerator)acceleratorparallel_devicescheckpoint_ioprecision_plugindebug_returnc                    s>   t sttt t j||t ||dd |  || _d| _d S )Nfork)r(   r)   cluster_environmentr*   r+   Zstart_methodF)r   ModuleNotFoundErrorstrsuper__init__r   r,   	_launched)selfr(   r)   r*   r+   r,   r-   	__class__ o/var/www/html/stable-diffusion-webui/venv/lib/python3.9/site-packages/pytorch_lightning/strategies/tpu_spawn.pyr4   8   s    	zTPUSpawnStrategy.__init__)r.   c                 C   s0   | j d u rt | _ nt| j tr*t | j _| j S N)_checkpoint_ior   
isinstancer   r*   r6   r9   r9   r:   r*   O   s
    


zTPUSpawnStrategy.checkpoint_io)ior.   c                 C   s
   || _ d S r;   )r<   )r6   r?   r9   r9   r:   r*   X   s    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   )r5   RuntimeErrortorch_xla.core.xla_modelcore	xla_modelZ
xla_device)r6   xmr9   r9   r:   root_device\   s    zTPUSpawnStrategy.root_devicec                 C   s   | j d ur| j  S dS Nr   )r0   
local_rankr>   r9   r9   r:   rG   d   s    zTPUSpawnStrategy.local_rank)dataloadersr.   c                 C   s(   t d ddd}t| tttf|d d S )N
dataloaderr.   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   r    )rJ   r9   r9   r:   check_has_lenj   s    z<TPUSpawnStrategy._validate_dataloader.<locals>.check_has_len)dtypeZwrong_dtypefunction)r   r
   objectr   r   )rH   rK   r9   r9   r:   _validate_dataloaderh   s    z%TPUSpawnStrategy._validate_dataloaderzpl.LightningModule)modelr.   c                 C   sd   | j j}|j|j|j|jf}|D ]>}| s |jdus:J t|jt	j
t	jfrRJ t|j q dS )zFValidate and fail fast if the dataloaders were passed directly to fit.N)trainerZ_data_connectorZ_train_dataloader_sourceZ_val_dataloader_sourceZ_test_dataloader_sourceZ_predict_dataloader_sourceZ	is_moduleinstancer=   plZLightningModuleZLightningDataModuler'   rO   )rP   Z	connectorsourcessourcer9   r9   r:   _validate_patched_dataloaderss   s    z.TPUSpawnStrategy._validate_patched_dataloadersc                    s8   t | dd lm  m} |t|| _t 	|S rF   )
r'   rV   Z)torch_xla.distributed.xla_multiprocessingZdistributedZxla_multiprocessingZMpModelWrapperr   wrapped_modelr3   connect)r6   rP   xmpr7   r9   r:   rX      s    
zTPUSpawnStrategy.connectc                 C   s   t | | _d S r;   )r   Z	_launcherr>   r9   r9   r:   _configure_launcher   s    z$TPUSpawnStrategy._configure_launcherz
pl.Trainer)rQ   r.   c                 C   s   | j s
J | j | | jr&dtjd< | js0J t| j}|   t| j| | 	  |j
jtjkr|| | t| j| j d S )N1PT_XLA_DEBUG)r(   setupr,   osenvironZlightning_moduler   model_to_devicer   Zsetup_precision_pluginstatefnr   ZFITTINGZsetup_optimizersr   Z
optimizersrE   )r6   rQ   Zshared_paramsr9   r9   r:   r]      s    




zTPUSpawnStrategy.setupc                 C   s   |S r;   r9   )r6   rP   r9   r9   r:   _setup_model   s    zTPUSpawnStrategy._setup_modelc                 C   s   t | j| jdS )N)Znum_replicasrank)dict
world_sizeglobal_rankr>   r9   r9   r:   distributed_sampler_kwargs   s    z+TPUSpawnStrategy.distributed_sampler_kwargsc                 C   s(   dd l m  m} |jtjv o&| jdkS Nr      )torch_xla.core.xla_env_varsrB   xla_env_varsZHOST_WORLD_SIZEr^   r_   rf   r6   Zxenvr9   r9   r:   is_distributed   s    zTPUSpawnStrategy.is_distributedr&   rI   c                 C   sN   t | ddlm} t||r$|S ||| j}|jj|_t|jdd |_	|S )Nr   r%   batch_sampler)
r'   rO   %torch_xla.distributed.parallel_loaderr&   r=   rE   Z_loaderZdatasetgetattrro   )r6   rJ   r&   r9   r9   r:   process_dataloader   s    


z#TPUSpawnStrategy.process_dataloaderc                 C   s   d S r;   r9   r>   r9   r9   r:   configure_ddp   s    zTPUSpawnStrategy.configure_ddpc                 C   s   | j | j| _d S r;   )rW   torE   rP   r>   r9   r9   r:   r`      s    z TPUSpawnStrategy.model_to_device)nameargskwargsr.   c                 O   s&   | j r"dd lm  m} || d S rF   )rn   rA   rB   rC   Z
rendezvous)r6   ru   rv   rw   rD   r9   r9   r:   barrier   s    zTPUSpawnStrategy.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)devicerL   r   )rn   r?   BytesIOtorchsave	bytearray	getbuffertensorrE   floatrA   rB   rC   
all_gathercpubytenumpyload)r6   ry   rz   bufferdataZdata_tensorrD   r9   r9   r:   	broadcast   s    

zTPUSpawnStrategy.broadcast)outputgroup	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)r{   )summeanavgzbCurrently, the TPUSpawnStrategy only supports `sum`, `mean`, `avg` for the reduce operation, got: r   reduce)r   r   )r=   r   r}   r   rE   r   ZSUMr2   lower
ValueErrorrA   rB   rC   Zmesh_reducer   rf   )r6   r   r   r   Zinvalid_reduce_opZinvalid_reduce_op_strrD   r9   r9   r:   r      s    

zTPUSpawnStrategy.reducec                 C   s   d| _ |   | jt_d S )NT)r5   set_world_ranksrg   r!   rd   r>   r9   r9   r:   setup_distributed   s    z"TPUSpawnStrategy.setup_distributedc                 C   s   | j d u rd S | j  t_d S r;   )r0   rg   r!   rd   r>   r9   r9   r:   r      s    
z TPUSpawnStrategy.set_world_ranks)rv   rw   r.   c                 O   sL   | j d usJ | j   | j |i |W  d    S 1 s>0    Y  d S r;   )rP   r+   Zval_step_contextr6   rv   rw   r9   r9   r:   validation_step   s    z TPUSpawnStrategy.validation_stepc                 O   sL   | j d usJ | j   | j |i |W  d    S 1 s>0    Y  d S r;   )rP   r+   Ztest_step_contextr   r9   r9   r:   	test_step   s    zTPUSpawnStrategy.test_stepc                 O   sL   | j d usJ | j   | j |i |W  d    S 1 s>0    Y  d S r;   )rP   r+   Zpredict_step_contextr   r9   r9   r:   predict_step   s    zTPUSpawnStrategy.predict_step)r   r.   c                 C   s   |    |S r;   _pod_progress_bar_force_stdoutr6   r   r9   r9   r:   training_step_end  s    z"TPUSpawnStrategy.training_step_endc                 C   s   |    |S r;   r   r   r9   r9   r:   validation_step_end	  s    z$TPUSpawnStrategy.validation_step_endc                 C   s   |    |S r;   r   r   r9   r9   r:   test_step_end  s    zTPUSpawnStrategy.test_step_endc                 C   s<   dd l m  m} | jdkr8tt|jddkr8t  d S ri   )	rk   rB   rl   rg   intr^   getenvZ
TPUVM_MODEprintrm   r9   r9   r:   r     s     z/TPUSpawnStrategy._pod_progress_bar_force_stdout)
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
        )r   N)r*   save_checkpoint)r6   r   r   r   r9   r9   r:   r     s    z TPUSpawnStrategy.save_checkpoint)r   r.   c                 C   s   | j dkr| j| dS )zpRemove checkpoint filepath from the filesystem.

        Args:
            filepath: Path to checkpoint
        r   N)rG   r*   remove_checkpoint)r6   r   r9   r9   r:   r   (  s    
z"TPUSpawnStrategy.remove_checkpoint)r   r   
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)
r=   r   ZdimZ	unsqueezeZtorch_xla.core.functionsrB   Z	functionsrA   rC   r   )r6   r   r   r   ZxfrD   r9   r9   r:   r   1  s
    

zTPUSpawnStrategy.all_gatherc                    s   t    tjdd  d S )Nr\   )r3   teardownr^   r_   popr>   r7   r9   r:   r   C  s    
zTPUSpawnStrategy.teardown)strategy_registryr.   c                 C   s.   |j d| ddd |j | j| | jj d d S )NZtpu_spawn_debugz&TPUSpawn Strategy with `debug` as TrueT)descriptionr,   )r   )registerstrategy_namer8   __name__)clsr   r9   r9   r:   register_strategiesG  s    z$TPUSpawnStrategy.register_strategies)NNNNF)N)r   )NN)N)NF)?r   
__module____qualname____doc__r   r   r   r}   r{   r   r   boolr   r4   propertyr*   setterrE   r   rG   staticmethodr	   r$   r"   rO   rV   rX   rZ   r]   r   rc   r   r2   rh   rn   r   rr   rs   r`   rx   r   r   r   r   r   r   r   r#   r   r   r   r   r   r   r   r   r   r   r   r   classmethodr   __classcell__r9   r9   r7   r:   r'   2   s        
  	r'   )Ar?   r^   typingr   r   r   r   r   r   r   r	   r}   Z#lightning_utilities.core.apply_funcr
   r   Ztorch.nnr   Ztorch.utils.datar   Zpytorch_lightningrS   Z!lightning_fabric.accelerators.tpur   Zlightning_fabric.pluginsr   r   Z%lightning_fabric.plugins.environmentsr   Zlightning_fabric.utilities.datar   Z$lightning_fabric.utilities.optimizerr   Z lightning_fabric.utilities.typesr   r   Zpytorch_lightning.overridesr   Z$pytorch_lightning.plugins.io.wrapperr   Z#pytorch_lightning.plugins.precisionr   Z&pytorch_lightning.strategies.ddp_spawnr   Z*pytorch_lightning.strategies.launchers.xlar   Z%pytorch_lightning.strategies.strategyr   Z3pytorch_lightning.trainer.connectors.data_connectorr   Z pytorch_lightning.trainer.statesr   Zpytorch_lightning.utilitiesr   r   Z&pytorch_lightning.utilities.exceptionsr    Z%pytorch_lightning.utilities.rank_zeror!   Z!pytorch_lightning.utilities.typesr"   r#   r$   rp   r&   r'   r9   r9   r9   r:   <module>   s<   (