a
    d%                     @   s  U d Z ddlmZ ddlmZ ddlmZ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mZ ddlmZ dd	lm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!m"Z" ddl#m$Z$m%Z%m&Z& ddl'm(Z(m)Z) dgZ*ee+ e,d< ee Z-eee-f Z.e
ee/ e/ef Z0erTee	d  Z1ee
e2ee
def e0df f  Z3neZ1eZ3eeddddZ4ee$e$ddddZ5ee$e$ddddZ6e7e7eee
e7e7f   dddZ8G dd dZ9dS ) z!The pipeline parallelism of Pipe.    )Queue)TracebackType)	TYPE_CHECKINGIterableListOptionalTupleTypeUnioncastSequenceN)Tensornn)record_function   )Checkpointing)CopyWait)forkjoin)Batch)
SkipLayout)SkipTrackerThroughPotalsuse_skip_tracker)AbstractStreamcurrent_stream
use_device)Taskcreate_workersPipeline__all__r   )	fork_fromjoin_toreturnc                 C   s:   |   }|  }t| | \| |< }t|| |||< d S N)Zfind_tensor_idxr   r   )r!   r"   Zfork_from_idxZjoin_to_idxZphony r%   q/var/www/html/stable-diffusion-webui/venv/lib/python3.9/site-packages/torch/distributed/pipeline/sync/pipeline.py_depend,   s    r'   )batchprev_streamnext_streamr#   c                 C   s:   t j||g| R  | d d < tdd | D | d d < d S )Nc                 S   s*   g | ]"}t |r"| s"| n|qS r%   torchZ	is_tensorZis_floating_pointdetach.0xr%   r%   r&   
<listcomp>7       z_copy.<locals>.<listcomp>)r   applytupler(   r)   r*   r%   r%   r&   _copy4   s    r6   c                 C   s:   t j||g| R  | d d < tdd | D | d d < d S )Nc                 S   s*   g | ]"}t |r"| s"| n|qS r%   r+   r.   r%   r%   r&   r1   =   r2   z_wait.<locals>.<listcomp>)r   r3   r4   r5   r%   r%   r&   _wait:   s    r7   )mnr#   c                 #   sL   t | | d D ]6  fddt td  |  dtd  |D V  qdS )z)Generates schedules for each clock cycle.r   c                    s   g | ]} | |fqS r%   r%   )r/   jkr%   r&   r1   P   r2   z!_clock_cycles.<locals>.<listcomp>r   N)rangemaxmin)r8   r9   r%   r;   r&   _clock_cycles@   s    r@   c                   @   s   e Zd ZdZeej eej eee	  e
eddddZee ddddZee eeeef  ee dd	d
dZee eeeef  ee dd	ddZdS )r   z"The pipeline parallelism for Pipe.N)
partitionsdevicescopy_streamsskip_layoutcheckpoint_stopr#   c                 C   s2   || _ || _|| _|| _|| _t|\| _| _d S r$   )rA   rB   rC   rD   rE   r   	in_queues
out_queues)selfrA   rB   rC   rD   rE   r%   r%   r&   __init__V   s    zPipeline.__init__)batchesr#   c                    sd   | j }| j}| j t|}t|} fdd|D }t||D ] }| ||| | ||| q>dS )zURuns pipeline parallelism.

        It modifies the given batches in place.

        c                    s   g | ]}t  qS r%   )r   )r/   _rD   r%   r&   r1   r   r2   z Pipeline.run.<locals>.<listcomp>N)rA   rB   rD   lenr@   fencecompute)rH   rJ   rA   rB   r8   r9   skip_trackersscheduler%   rL   r&   rune   s    zPipeline.run)rJ   rQ   rP   r#   c              	   C   s   | j }| j}|D ]\}}|dkr>|dkr>t||d  ||  || | }||D ]0\}	}
}||	 | }|| || |||
| qT|dkr||d  | }t|| || qdS )zWCopies micro-batches after computation for the previous
        micro-batches.
        r   r   N)rC   rD   r'   Zcopy_policycopyr6   )rH   rJ   rQ   rP   rC   rD   ir:   r*   Zprev_jnsnamer)   r%   r%   r&   rN   x   s    zPipeline.fencec              	   C   s  | j }| j}| j}| j}| j d js(d}t|}dd |D }	d}
|D ]\}}|| }|| }|dkr~t||| | |	|  ||k }|r||| ||dtjt	t
t
tddd}t||}t|	| |j|jd	}~~n>|||| ||fttjt	t
t
td
dd}t|	| |dd	}~| j| | qF|D ]\}}| j|  \}}|
durVq,n|sjtt|}
q,ttttf |\}}||d krt||	| || |  t||  || W d   n1 s0    Y  |||< q,|
dur|
d |
d |
d dS )z0Runs tasks with synchronization to copy streams.r   c                 S   s   g | ]}t |qS r%   )r   )r/   dr%   r%   r&   r1      r2   z$Pipeline.compute.<locals>.<listcomp>N)	partitionskip_trackerchunk_idpart_id)rX   rY   rZ   r[   r#   c              	   W   sr   t |V td||f & | | W  d    W  d    S 1 sF0    Y  W d    n1 sd0    Y  d S Nzchunk%d-part%d)r   r   )rX   rY   rZ   r[   inputsr%   r%   r&   function   s    z"Pipeline.compute.<locals>.function)rO   finalize)r(   rX   rY   rZ   r[   r#   c              	   S   st   t |X td||f ( | |W  d    W  d    S 1 sH0    Y  W d    n1 sf0    Y  d S r\   )r   r   call)r(   rX   rY   rZ   r[   r%   r%   r&   rO      s    z!Pipeline.compute.<locals>.computer      )rA   rB   rC   rE   ZtrainingrM   r7   r   Moduler   intTensorOrTensorsr   r   
checkpointZ	recomputer   rF   putrG   getr   ExcInfor   r   r_   with_traceback)rH   rJ   rQ   rP   rA   rB   rC   rE   r9   Zstreamsexc_inforT   r:   r(   rX   re   r^   ZchkZtaskrO   okpayloadr%   r%   r&   rO      sz    




*
zPipeline.compute)__name__
__module____qualname____doc__r   r   Z
Sequentialr,   Zdevicer   r   rc   rI   r   rR   r   r   rN   rO   r%   r%   r%   r&   r   S   s   
):rp   queuer   typesr   typingr   r   r   r   r   r	   r
   r   r   r,   r   r   Ztorch.autograd.profilerr   re   r   rS   r   r   
dependencyr   r   Z
microbatchr   Zskip.layoutr   Zskip.trackerr   r   streamr   r   r   Zworkerr   r   r    str__annotations__ZTensorsrd   BaseExceptionrh   ZInQueueboolZOutQueuer'   r6   r7   rc   r@   r   r%   r%   r%   r&   <module>   s8   ,$"