a
    df;                     @   s   d dl Z d dlZd dlZd dlZd dlmZmZmZmZm	Z	m
Z
 d dlZd dlmZ d dlZd dlmZ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rd dlZndZe e!Z"G dd deZ#G dd dZ$dS )    N)AnyCallableDictListOptionalUnion)Tensor)LRSchedulerReduceLROnPlateau)Strategy
TBroadcast)extract_batch_size)MisconfigurationException)_HIVEMIND_AVAILABLE)is_overridden)rank_zero_warnc                       s  e Zd ZU dZeed< d6eeee eee eee eee	 e
e
eee ee eeeef  ed fd	d
ZddddZeedddZeejdddZeedddZeedddZddd fddZddddZddddZdddd d!Zd7eeedd#d$d%Zeeef eeeeef d&d'd(Zd8eee eed)d*d+Zddd,d-Z eedd.d/d0Z!d9e"ee"d1d2d3Z#dd fd4d5Z$  Z%S ):HivemindStrategyZPL_INITIAL_PEERSINITIAL_PEERS_ENVlightning_runNF      @      >@)target_batch_sizerun_id
batch_sizedelay_state_averagingdelay_optimizer_stepdelay_grad_averagingoffload_optimizerreuse_grad_buffersscheduler_fnmatchmaking_timeaveraging_timeoutverboseaverager_optshost_maddrsinitial_peersoptimizer_kwargsc                    s  t rt dkrtdt   || _|| _|| _|	| _	|pD|pD|| _
d| _d| _|| _|| _tf |
||||||dur||ntdd||d	|| _|   tjd||dur|ndd	gd
| _dd | j D }|du rtdd| dd| d d| _dS )a  Provides capabilities to train using the Hivemind Library, training collaboratively across the internet
        with unreliable machines. For more information, `refer to the docs <https://pytorch-
        lightning.readthedocs.io/en/latest/strategies/hivemind.html>`__.

        .. warning:: ``HivemindStrategy`` is experimental and subject to change.

        Arguments:

            target_batch_size: When training, the batch size to accumulate to before running a step. The larger this
                batch size, the more work can be done asynchronously without communication.

            run_id: A unique identifier of this training run, used as a common prefix for all DHT keys.
                See ``https://learning-at-home.readthedocs.io/en/latest/user/dht.html``.

            batch_size: The local batch size per process. If not provided, we infer this from the first batch of data
                passed in at training (lazy). Note that this should not change throughout training.

            delay_state_averaging: If enabled (default), average parameters and extra tensors in a background thread;
                if set to False, average parameters synchronously within the
                corresponding :meth:`hivemind.Optimizer.step` call.

            delay_optimizer_step: Run optimizer in background, apply results in future .step. requires
                :paramref:`~pytorch_lightning.strategies.hivemind.HivemindStrategy.offload_optimizer`.

            delay_grad_averaging: Average gradients in background; requires
                :paramref:`~pytorch_lightning.strategies.hivemind.HivemindStrategy.offload_optimizer` and
                :paramref:`~pytorch_lightning.strategies.hivemind.HivemindStrategy.delay_optimizer_step`.

            offload_optimizer: Offload the optimizer to host memory, saving GPU memory for parameters and gradients.

            reuse_grad_buffers: Use the model's gradient buffers (params.grad) for gradient accumulation
                which is more memory efficient. Lightning will automatically disable ``zero_grad``
                in the ``LightningModule``.

            scheduler_fn: callable(optimizer) -> PyTorch LRScheduler or a pre-initialized PyTorch scheduler.
                When using `offload_optimizer`/`delay_optimizer_step`/`delay_state_averaging` ``scheduler_fn``
                is required to be passed to the ``HivemindStrategy``. This is because the optimizer
                is re-created and the scheduler needs to be re-created as well.

            matchmaking_time: When looking for group, wait for peers to join for up to this many seconds.
                Increase if you see "averaged gradients with N peers" where N is below 0.9x on >=25% of epochs.
                Training with low-latency network, decreasing matchmaking_time allows training with smaller batch sizes.

            averaging_timeout: If an averaging step hangs for this long, it will be cancelled automatically.
                Increase averaging_timeout if you see "Proceeding with local gradients" at least 25% of the time.
                Do not set this timeout too high, as it may cause your optimizer to hang
                after some types of network errors.

            verbose: Report internal Hivemind events such as accumulating gradients and running background tasks.

            averager_opts: Additional keyword arguments forwarded to both
                ``GradientAverager`` and ``TrainingStateAverager``.

            host_maddrs: List of multi-addrs to create visible peers for other processes.
                `https://learning-at-home.readthedocs.io/en/latest/user/dht.html#running-across-the-internet`

            initial_peers: If connecting to a running process, a list of initial peers needs to be passed in.
                This can also be set via the env variable ``INITIAL_PEERS``.

            **optimizer_kwargs: kwargs are passed to the :class:`hivemind.Optimizer` class.
        LinuxzTo use the `HivemindStrategy`, you must have Hivemind installed and be running on Linux. Install it by running `pip install -U hivemind`.Ng      ?)request_timeout)	r    r!   r   r   r   r   r#   r"   r   Tz/ip4/0.0.0.0/tcp/0z/ip4/0.0.0.0/udp/0/quic)startr%   r$   c                 S   s(   g | ] }t | d  jst|qS )r   )	ipaddress
ip_addressvaluesis_loopbackstr).0a r1   n/var/www/html/stable-diffusion-webui/venv/lib/python3.9/site-packages/pytorch_lightning/strategies/hivemind.py
<listcomp>   s   z-HivemindStrategy.__init__.<locals>.<listcomp>zD
Other machines can connect running the same command:
INITIAL_PEERS=,zR python ...
or passing the peers to the strategy:
HivemindStrategy(initial_peers='z')F)r   platformsystemr   super__init___initial_peers_target_batch_size_batch_size_scheduler_fn_require_scheduler_fn_opt_optimizer_zero_grad_original_run_id_reuse_grad_buffersdict_optimizer_kwargs_parse_env_initial_peershivemindZDHTdhtZget_visible_maddrsloginfojoin_hivemind_initialized)selfr   r   r   r   r   r   r   r   r   r    r!   r"   r#   r$   r%   r&   Zvisible_addresses	__class__r1   r2   r8      sZ    P


zHivemindStrategy.__init__returnc                 C   s2   t j| j| j}t|tr&|dn| j| _d S )Nr4   )osenvirongetr   r9   
isinstancer.   split)rK   r%   r1   r1   r2   rD      s    z)HivemindStrategy._parse_env_initial_peersc                 C   s   | j r| j jjjS dS )N   )r>   trackerZglobal_progress	num_peersrK   r1   r1   r2   rW      s    zHivemindStrategy.num_peersc                 C   sj   ddl m} ddlm} t| j|r:tdtj	  S t| j|rPtdS t
d| jjj dd S )Nr   )CPUAccelerator)CUDAAcceleratorzcuda:cpuz6Was unable to infer device type from the accelerator: .)Z"pytorch_lightning.accelerators.cpurY   Z#pytorch_lightning.accelerators.cudarZ   rS   ZacceleratortorchdeviceZcudaZcurrent_devicer   rM   __name__)rK   rY   rZ   r1   r1   r2   root_device   s    
zHivemindStrategy.root_devicec                 C   s   dS )Nr   r1   rX   r1   r1   r2   global_rank   s    zHivemindStrategy.global_rankc                 C   s   dS )NTr1   rX   r1   r1   r2   is_global_zero   s    zHivemindStrategy.is_global_zeroz
pl.Trainer)trainerrO   c                    s0   |    t | | jjdkr,t | j_d S )NZ16)model_to_devicer7   setupZprecision_pluginZ	precisionrE   Z
GradScalerZscaler)rK   rc   rL   r1   r2   re      s    zHivemindStrategy.setupc              
   C   s   t | jdkrtd| jd }| jr8| jd u r8td | jrD| jnd }| jrT|jnd }| jrft|n|}tj	f | j
| j|||| j| jd| j}| js| | |  |g| _|| _| jr| jd usJ | jj| _|   d S )NrU   z3Hivemind only supports training with one optimizer.r   zEnabling `delay_optimizer_step`, `delay_state_averaging` or `offload_optimizer` requires a `scheduler_fn` to be passed to the strategy if a scheduler is being used (this is because the optimizer is re-created within Hivemind).)rF   r   params	optimizer	schedulerr   Zbatch_size_per_step)lenZ
optimizersr   r=   r<   r   Zparam_groupstyperE   Z	OptimizerrF   r@   r:   r;   rC   _wrap_schedulersZload_state_from_peersr>   rA   lightning_moduleoptimizer_zero_gradr?   _disable_zero_grad)rK   rg   rh   rf   optr1   r1   r2   _initialize_hivemind   s<    


z%HivemindStrategy._initialize_hivemindc                 C   s:   | j }td|r$|d usJ td |d us0J d |_d S )Nrm   zYou have overridden `optimizer_zero_grad` which will be disabled. When `HivemindStrategy(reuse_grad_buffers=True)`, the optimizer cannot call zero grad, as this would delete the gradients before they are averaged.)rl   r   r   rm   )rK   rl   r1   r1   r2   rn      s    
z#HivemindStrategy._disable_zero_gradhivemind.Optimizer)ro   rO   c                 C   sB   | j D ]6}|j}t|tr.td| jj dt||d|_qd S )NzCThe `ReduceLROnPlateau` scheduler is not currently supported with `z`.)rg   rh   )Zlr_scheduler_configsrh   rS   r
   
ValueErrorrM   r_   HiveMindScheduler)rK   ro   Zscheduler_configrh   r1   r1   r2   rk      s    

z!HivemindStrategy._wrap_schedulersr   )batch	batch_idxdataloader_idxrO   c              
   C   sv   | j srd| _ | jd u rjz t|| _td| j  W n2 ttfyh } ztd|W Y d }~n
d }~0 0 |   d S )NTz;Found per machine batch size automatically from the batch: zWe tried to infer the batch size from the first batch of data. Please provide the batch size to the Strategy by ``Trainer(strategy=HivemindStrategy(batch_size=x))``. )rJ   r;   r   rG   rH   r   RecursionErrorrp   )rK   rt   ru   rv   er1   r1   r2   on_train_batch_start  s    

z%HivemindStrategy.on_train_batch_start)tensorargskwargsrO   c                 O   s   |S Nr1   )rK   rz   r{   r|   r1   r1   r2   reduce  s    zHivemindStrategy.reduce)rz   group
sync_gradsrO   c                 C   s   |S r}   r1   )rK   rz   r   r   r1   r1   r2   
all_gather  s    zHivemindStrategy.all_gatherc                 C   s    | j d usJ | j | j d S r}   )modeltor`   rX   r1   r1   r2   rd     s    z HivemindStrategy.model_to_device)r{   r|   rO   c                 O   s   d S r}   r1   )rK   r{   r|   r1   r1   r2   barrier  s    zHivemindStrategy.barrier)objsrcrO   c                 C   s   |S r}   r1   )rK   r   r   r1   r1   r2   	broadcast"  s    zHivemindStrategy.broadcastc                    sP   | j d ur| jd ur| j | j_| jr.| j  td | j  t 	  d S )NzShutting down hivemind DHT.)
r?   rl   rm   r>   shutdownrG   rH   rF   r7   teardownrX   rL   r1   r2   r   %  s    



zHivemindStrategy.teardown)r   NFNFNFNr   r   FNNN)r   )NF)r   )&r_   
__module____qualname__r   r.   __annotations__intr   boolr   floatr   r   r   r   r8   rD   propertyrW   r]   r^   r`   ra   rb   re   rp   rn   rk   ry   r   r~   r   rd   r   r   r   r   __classcell__r1   r1   rL   r2   r      sl   
               %$r   c                   @   sd   e Zd ZU dZee ed< deddddZde	e
 ddd	d
ZeddddZedddZdS )rs   zWrapper for schedulers to prevent Lightning from stepping the scheduler too soon.

    This code ensures that we only step when the HiveMind optimizer reaches the global step.
    Zbase_lrsrq   N)rg   rh   rO   c                 C   s,   dd |j  D | _ || _|| _d| _d S )Nc                 S   s   i | ]\}}|d vr||qS ))step__del__r1   )r/   kvr1   r1   r2   
<dictcomp>=      z.HiveMindScheduler.__init__.<locals>.<dictcomp>)__dict__itemsrg   rh   current_step)rK   rg   rh   r1   r1   r2   r8   :  s    zHiveMindScheduler.__init__)epochrO   c                 C   s0   | j | jjk r,| jj|d |  j d7  _ q d S )N)r   rU   )r   rg   Zlocal_epochrh   r   )rK   r   r1   r1   r2   r   C  s    zHiveMindScheduler.step)
state_dictrO   c                 C   s   | j | d S r}   )rh   load_state_dict)rK   r   r1   r1   r2   r   H  s    z!HiveMindScheduler.load_state_dictrN   c                 C   s
   | j  S r}   )rh   r   rX   r1   r1   r2   r   K  s    zHiveMindScheduler.state_dict)N)r_   r   r   __doc__r   r   r   r	   r8   r   r   r   r   r   r   r1   r1   r1   r2   rs   2  s   
	rs   )%r*   loggingrP   r5   typingr   r   r   r   r   r   r]   r   Zpytorch_lightningplZ lightning_fabric.utilities.typesr	   r
   Z%pytorch_lightning.strategies.strategyr   r   Z pytorch_lightning.utilities.datar   Z&pytorch_lightning.utilities.exceptionsr   Z#pytorch_lightning.utilities.importsr   Z)pytorch_lightning.utilities.model_helpersr   Z%pytorch_lightning.utilities.rank_zeror   rE   	getLoggerr_   rG   r   rs   r1   r1   r1   r2   <module>   s,    

  