a
    d3                     @   s  d dl Z d dlZd dl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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% e
dZ&e&rd dl'm(Z) d dl*m+Z+ d dl,m-Z- d dl.m/Z/m0Z0m1Z1m2Z2 d d	l.mZ3 d dl4m5Z6 ej7e3j7ej8e3j8ej9e3j9ej:e3j:ej;e3j;ej<e3j<ej=e3j=de3j>de3j>de3j7i
Z?ni Z?e @eAZBG dd deZCG dd deZDdS )    N)AnyDictListOptionalUnion)module_available)Tensor)Module)CheckpointIOClusterEnvironment)_optimizers_to_device)
reset_seed)ReduceOp)_LightningModuleWrapperBase$_LightningPrecisionModuleWrapperBase)PrecisionPlugin)DDPStrategy)
TBroadcast)	TrainerFn)MisconfigurationExceptionzbagua.torch_api)	Algorithm)QAdamOptimizer)allreduce_inplacebarrierbroadcast_objectis_initialized)DistributedDataParallel_V1_9_0avgmeansumc                       sV   e Zd Zd	eedef  eedef  dd fddZeeed fddZ  Z	S )
LightningBaguaModuleNzpl.LightningModule)forward_module	pl_modulereturnc                    s<   |  || |p|}t j|d |jj t| | _d S )N)r!   )Z_validate_init_argumentssuper__init__	__class____name__idZ_bagua_module_name)selfr!   r"   r&    k/var/www/html/stable-diffusion-webui/venv/lib/python3.9/site-packages/pytorch_lightning/strategies/bagua.pyr%   A   s    zLightningBaguaModule.__init__)inputskwargsr#   c                    sb   | j }|j}|d urR|jr@| jj|i |}|js<d|jj_|S t	 j
|i |S | j|i |S )NF)lightning_moduleZ_trainertrainingZ_forward_moduleZtraining_stepautomatic_optimizationmodelinnerrequire_backward_grad_syncr$   forward)r)   r-   r.   r"   traineroutputr*   r+   r,   r5   L   s    
zLightningBaguaModule.forward)NN)
r'   
__module____qualname__r   r   r   r%   r   r5   __classcell__r+   r+   r*   r,   r    @   s     r    c                       sD  e Zd ZdZd-eeed eeej	  ee
 ee ee eeeeef f d fddZdd	d
dZdd	ddZdd	ddZdddddZdd	ddZdddddZeddddZeeddddZdd	 fdd Zdd	d!d"Zd.eeed$d%d&Zdd	d'd(Z d/e!ee eee"ef  e!d*d+d,Z#  Z$S )0BaguaStrategybaguagradient_allreduceTNzpl.accelerators.Accelerator)	algorithmflattenacceleratorparallel_devicescluster_environmentcheckpoint_ioprecision_pluginbagua_kwargsc           	         s8   t stdt j|||||d || _|| _|| _dS )a  Strategy for training using the `Bagua <https://github.com/BaguaSys/bagua>`_ library, with advanced
        distributed training algorithms and system optimizations.

        This strategy requires the `bagua` package to be installed. See
        `installation guide <https://tutorials.baguasys.com/installation>`_ for more information.

        The :class:`BaguaStrategy` is only supported on GPU and on Linux systems.

        Arguments:
            algorithm: Distributed algorithm used to do the actual communication and update. Built-in algorithms
                include "gradient_allreduce", "bytegrad", "decentralized", "low_precision_decentralized", "qadam" and
                "async".
            flatten: Whether to flatten the Bagua communication buckets. The flatten operation will reset data
                pointer of bucket tensors so that they can use faster code paths.
            bagua_kwargs: Additional keyword arguments that will be passed to initialize the Bagua algorithm. More
                details on keyword arguments accepted for each algorithm can be found in the
                `documentation <https://bagua.readthedocs.io/en/latest/autoapi/bagua/torch_api/algorithms/index.html>`_.
        zcTo use the `BaguaStrategy`, you must have `Bagua` installed. Use `pip install bagua` to install it.)r@   rA   rB   rC   rD   N)_BAGUA_AVAILABLEr   r$   r%   _bagua_algorithm_bagua_flatten_bagua_kwargs)	r)   r>   r?   r@   rA   rB   rC   rD   rE   r*   r+   r,   r%   e   s    zBaguaStrategy.__init__)r#   c                 C   s   t   |   |   d S N)r   Zset_world_ranks_init_bagua_distributedr)   r+   r+   r,   setup_distributed   s    zBaguaStrategy.setup_distributedc                 C   sN   |    td| j d| jd  d| j  tj| j t	 sJt
  d S )Nz-Initializing Bagua Distributed: GLOBAL_RANK: z
, MEMBER:    /)_set_node_environment_variablesloginfoglobal_rank
world_sizetorchZcudaZ
set_device
local_rankr   r<   Zinit_process_grouprL   r+   r+   r,   rK      s    z%BaguaStrategy._init_bagua_distributedc                 C   sd   | j jtjd< t| j jtjd< t| jtjd< t| jtjd< t| jtjd< t| j	tjd< dS )a  Set the environment variables as required by the :func:`bagua.init_process_group` call.

        This enables the use of other cluster environments which don't set these exact variables, e.g., Bagua can be
        launched with ``torch.distributed.run``.
        ZMASTER_ADDRZMASTER_PORTZRANKZ	NODE_RANKZ
WORLD_SIZEZ
LOCAL_RANKN)
rB   Zmain_addressosenvironstrZ	main_portrS   Z	node_rankrT   rV   rL   r+   r+   r,   rP      s    z-BaguaStrategy._set_node_environment_variablesz
pl.Trainer)r6   r#   c                 C   s   |  | j| _|  r|   | jd us,J | j| |   |jj}|t	j
krn| jrn| jrn| j| j| _|   |t	j
kr| | t| j| j | | d S rJ   )	broadcastZ"_rank_0_will_call_children_scriptsZ_should_run_deadlock_detectionZ&_share_information_to_prevent_deadlockr@   setupZmodel_to_devicestatefnr   ZFITTINGZ_layer_syncr2   applyZsetup_precision_pluginZsetup_optimizersr   
optimizersZroot_device_configure_bagua_model)r)   r6   Z
trainer_fnr+   r+   r,   r[      s    


zBaguaStrategy.setupc                 C   sP   t dd | jD }|r4t| jdks4t| jdkr<td| jd | jd< d S )Nc                 S   s   g | ]}t |tqS r+   )
isinstancer   ).0optr+   r+   r,   
<listcomp>       z8BaguaStrategy._check_qadam_optimizer.<locals>.<listcomp>rN   zDBagua QAdam can only accept one QAdamOptimizer and one LR Scheduler.r   Zq_adam_optimizer)anyr_   lenZlr_scheduler_configsr   rI   )r)   Zhas_qadam_optimizerr+   r+   r,   _check_qadam_optimizer   s     z$BaguaStrategy._check_qadam_optimizerc                 C   s:   t | j}| || _|jr6| jdkr6| jj| j d S Nasync)r    r2   _setup_modelr0   rG   bagua_algorithmresume)r)   r6   r2   r+   r+   r,   r`      s    
z$BaguaStrategy._configure_bagua_modelBaguaDistributedDataParallel)r2   r#   c                 C   s<   | j dkr|   tj| j fi | j}t|| j|| jdS )z0Wraps the model into a Bagua distributed module.Zqadam)moduler_   r>   Zgradient_as_bucket_view)rG   rh   r   initrI   rn   r_   rH   )r)   r2   r>   r+   r+   r,   rk      s    
zBaguaStrategy._setup_model)strategy_registryr#   c                 C   s   |j | j| | jj d d S )N)description)registerstrategy_namer&   r'   )clsrq   r+   r+   r,   register_strategies   s
    z!BaguaStrategy.register_strategiesc                    sT   | j d usJ | j jjr2| jdkr2| jj| j t| jtrF| j | _t	 
  d S ri   )r/   r6   r0   rG   r2   rl   abortra   rn   r$   teardownrL   r*   r+   r,   rx      s    zBaguaStrategy.teardownc                 O   s   t  rt  d S rJ   )r   r   )r)   argsr.   r+   r+   r,   r     s    zBaguaStrategy.barrierr   )objsrcr#   c                 C   s
   t ||S rJ   )r   )r)   rz   r{   r+   r+   r,   rZ     s    zBaguaStrategy.broadcastc                 C   s$   | j d usJ | j js d| jj_d S )NT)r/   r1   r2   r3   r4   rL   r+   r+   r,   post_training_step  s    z BaguaStrategy.post_training_stepr   )tensorgroup	reduce_opr#   c                 C   s`   t |ts|S |durtd|du r.tj}n"t|d}|du rPtd| t||d |S )a  Reduces a tensor from several distributed processes to one aggregated tensor.

        Args:
            tensor: The tensor to sync and reduce.
            group: The process group to gather results from. Defaults to all processes (world).
            reduce_op: The reduction operation.
                Can also be a string 'sum' or ReduceOp.

        Return:
            The reduced value, except when the input was not a tensor the output remains is unchanged.
        NzW`Bagua` does not support allreduce using a subcommunicator at this time. Unset `group`.z.Unrecognized `reduce_op` for `BaguaStrategy`: )op)ra   r   
ValueErrorBaguaReduceOpAVG_bagua_reduce_opsgetr   )r)   r}   r~   r   r   r+   r+   r,   reduce  s    
zBaguaStrategy.reduce)r=   TNNNNN)r   )Nr   )%r'   r8   r9   rt   rY   boolr   r   rU   Zdevicer   r
   r   r   r   r   r%   rM   rK   rP   r[   rh   r`   r	   rk   classmethodrv   rx   r   r   intrZ   r|   r   r   r   r:   r+   r+   r*   r,   r;   b   sH          .	 r;   )EloggingrW   typingr   r   r   r   r   rU   Z lightning_utilities.core.importsr   r   Ztorch.nnr	   Zpytorch_lightningplZlightning_fabric.pluginsr
   r   Z$lightning_fabric.utilities.optimizerr   Zlightning_fabric.utilities.seedr   Z lightning_fabric.utilities.typesr   Z pytorch_lightning.overrides.baser   r   Z#pytorch_lightning.plugins.precisionr   Z pytorch_lightning.strategies.ddpr   Z%pytorch_lightning.strategies.strategyr   Z pytorch_lightning.trainer.statesr   Z&pytorch_lightning.utilities.exceptionsr   rF   Zbagua.torch_apiZ	torch_apir<   Zbagua.torch_api.algorithmsr   Z!bagua.torch_api.algorithms.q_adamr   Zbagua.torch_api.communicationr   r   r   r   r   Z)bagua.torch_api.data_parallel.distributedr   rn   ZSUMZPRODUCTZMINMAXBANDBORBXORr   r   	getLoggerr'   rQ   r    r;   r+   r+   r+   r,   <module>   sP   
"