U
    ©Ê¹j˜%  ã                   @   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%   úg/var/www/html/bwave/python/venv/lib/python3.8/site-packages/torch/distributed/pipeline/sync/pipeline.pyÚ_depend+   s    r'   )ÚbatchÚprev_streamÚnext_streamr#   c                 C   s8   t j||f| žŽ | 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>6   s     z_copy.<locals>.<listcomp>)r   ÚapplyÚtuple©r(   r)   r*   r%   r%   r&   Ú_copy3   s    r5   c                 C   s8   t j||f| žŽ | 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   <   s     z_wait.<locals>.<listcomp>)r   r2   r3   r4   r%   r%   r&   Ú_wait9   s    r6   )Ú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   O   s     z!_clock_cycles.<locals>.<listcomp>r   N)ÚrangeÚmaxÚmin)r7   r8   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$   )r@   rA   rB   rC   rD   r   Ú	in_queuesÚ
out_queues)Úselfr@   rA   rB   rC   rD   r%   r%   r&   Ú__init__U   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/   Ú_©rC   r%   r&   r1   q   s     z Pipeline.run.<locals>.<listcomp>N)r@   rA   rC   Úlenr?   ÚfenceÚcompute)rG   rI   r@   rA   r7   r8   Úskip_trackersÚscheduler%   rK   r&   Úrund   s    zPipeline.run)rI   rP   rO   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)rB   rC   r'   Zcopy_policyÚcopyr5   )rG   rI   rP   rO   rB   rC   Úir9   r*   Zprev_jÚnsÚnamer)   r%   r%   r&   rM   w   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k	�rV�q,n|�sjtt|ƒ}
�q,ttttf |ƒ\}}||d k�r¦t||	| || | ƒ t|| ƒ� | |¡ W 5 Q R X |||< �q,|
dk	�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   ž   s     z$Pipeline.compute.<locals>.<listcomp>N)Ú	partitionÚskip_trackerÚchunk_idÚpart_id)rW   rX   rY   rZ   r#   c                 W   sP   t |ƒ�> td||f ƒ�" | |Ž W  5 Q R £ W  5 Q R £ S Q R X W 5 Q R X d S ©Nzchunk%d-part%d)r   r   )rW   rX   rY   rZ   Úinputsr%   r%   r&   ÚfunctionÆ   s    z"Pipeline.compute.<locals>.function)rN   Úfinalize)r(   rW   rX   rY   rZ   r#   c                 S   sR   t |ƒ�@ td||f ƒ�$ |  |¡W  5 Q R £ W  5 Q R £ S Q R X W 5 Q R X d S r[   )r   r   Úcall)r(   rW   rX   rY   rZ   r%   r%   r&   rN   Ö   s    z!Pipeline.compute.<locals>.computer   é   )r@   rA   rB   rD   ZtrainingrL   r6   r   ÚModuler   ÚintÚTensorOrTensorsr   r   Ú
checkpointZ	recomputer   rE   ÚputrF   Úgetr   ÚExcInfor   r   r^   Úwith_traceback)rG   rI   rP   rO   r@   rA   rB   rD   r8   ÚstreamsÚexc_inforS   r9   r(   rW   rd   r]   ZchkÚtaskrN   ÚokÚpayloadr%   r%   r&   rN   �   sz    ûú

ûú



zPipeline.compute)Ú__name__Ú
__module__Ú__qualname__Ú__doc__r   r   Z
Sequentialr,   Zdevicer   r   rb   rH   r   rQ   r   r   rM   rN   r%   r%   r%   r&   r   R   s&   
ù  þ  þ):rq   Úqueuer   Útypesr   Útypingr   r   r   r   r   r	   r
   r   r   r,   r   r   Ztorch.autograd.profilerr   rd   r   rR   r   r   Ú
dependencyr   r   Z
microbatchr   Zskip.layoutr   Zskip.trackerr   r   Ústreamr   r   r   Zworkerr   r   r    ÚstrÚ__annotations__ZTensorsrc   ÚBaseExceptionrg   ZInQueueÚboolZOutQueuer'   r5   r6   rb   r?   r   r%   r%   r%   r&   Ú<module>   s8   ,$"