B
    F.\                 @   s  d dl mZmZmZ d dlmZ d dlmZ d dlZd dl	Z	d dl
mZ d dlmZmZmZmZmZ d dlmZ d dlmZmZmZmZ d d	lmZmZmZ ed
ddd Zed
ddd Zdd Ze dd Ze	jj dded
ddd Z!dd Z"e	jj#ej$d  dkddeed
dgd edddd  Z%ed
dd!d" Z&ed
dd#d$ Z'ed
dd%d& Z(ed
dd'd( Z)ed
dd)d* Z*ed
dd+d, Z+ed
dd-d. Z,dS )/    )print_functiondivisionabsolute_import)	timedelta)sleepN)gen)ClientQueueNannyworker_clientwait)time)gen_clusterincslowdiv)clientcluster_fixtureloopT)r   c       
   	   c   s   t dV }t dV }t dV }|j| ks,t| td}||V  ||V  | V }|j|jksjtt	t
j |jddV  W d Q R X ~~t
dV  |jst| V  t }	x(|jrt
dV  t |	d k stqW d S )Nxy   g?)timeoutg{Gz?   )r	   r   AssertionErrorsubmitr   putgetkeypytestraisesr   TimeoutErrorr   Ztasksr   )
csabr   r   xxfuturefuture2start r*   <lib/python3.7/site-packages/distributed/tests/test_queues.py
test_queue   s&    





r,   c          	   c   sn   t dV }t dV }|j| ks"t|dV  | V }|dksDtttj |jddV  W d Q R X d S )Nr   )r   Zhellog?)r   )	r	   r   r   r   r   r   r    r   r!   )r"   r#   r$   r%   r   r&   datar*   r*   r+   test_queue_with_data-   s    


r.   c             C   sf   |  dd d}td}td}|| | dks:t| dksJt| }| dksbtd S )Nc             S   s   | d S )Nr   r*   )r   r*   r*   r+   <lambda>=   s    ztest_sync.<locals>.<lambda>
   r   r      )r   r	   r   qsizer   r   result)r   r'   r   r&   r(   r*   r*   r+   	test_sync<   s    
r4   c       
      c   s   t | jddV }|dd d}tdV }||V  ~| V  tdV  t | jddV }tdV }| V }|V }	|	dkst	| V  d S )	NT)asynchronousc             S   s   | d S )Nr   r*   )r   r*   r*   r+   r/   K   s    z#test_hold_futures.<locals>.<lambda>r0   qg?r1   )
r   addressr   r	   r   closer   r   r   r   )
r#   r$   r%   Zc1r'   Zq1c2q2r(   r3   r*   r*   r+   test_hold_futuresH   s    



r;   z$getting same client from main thread)reasonc             #   s:   t    fdd}| |dV    V }|dks6td S )Nc                s     | d  d S )Nr   )r   )r   )r6   r*   r+   fa   s    ztest_picklability.<locals>.fr0   r1   )r	   r   r   r   )r"   r#   r$   r%   r=   r3   r*   )r6   r+   test_picklability\   s
    
r>   c                s6   t    fdd}| |d    dks2td S )Nc                s     | d  d S )Nr   )r   )r   )r6   r*   r+   r=   l   s    z!test_picklability_sync.<locals>.fr0   r1   )r	   r   r3   r   r   )r   r=   r*   )r6   r+   test_picklability_synci   s    r?      zMulti-client issues)z	127.0.0.1r@   r   )r   ZncoresZWorkerr   c       
      g   s   dd }t d| d}| tdV }x|D ]}||V  q*W | |td}| |V }tdd |D sptt|dkst|	 V }	|	rtd S )	Nc          	   S   sn   t  ^}td|d}x>tdD ]2}| }| }|t|}|| td qW |  }|S Q R X d S )Nr   )r   d   g{Gz?)	r   r	   ranger   r3   r   r   r   r   )ir"   r6   _r'   r   r   r3   r*   r*   r+   r=   y   s    
ztest_race.<locals>.fr   )r   r   c             s   s   | ]}|d kV  qdS )2   Nr*   ).0rr*   r*   r+   	<genexpr>   s    ztest_race.<locals>.<genexpr>i  )
r	   scatterrB   r   mapZgatherallr   sumr2   )
r"   r#   Zworkersr=   r6   Lr'   ZfuturesZresultsr2   r*   r*   r+   	test_racet   s    

rN   c       	      c   s   t d}| dV }xtdD ]}||V  qW |jd |jhksHtxRtdD ]F}| V }|jd |jhksvtt	dV  |jd |jhksRtqRW | V  t
 }x,|jd rt	dV  t
 | dk stqW d S )	Nr   {   r   zqueue-x   g?g{Gz?r@   )r	   rI   rB   r   Z
wants_whatr   r   r   r   r   r   )	r"   r#   r$   r%   r6   r'   rC   r(   r)   r*   r*   r+   test_same_futures   s    

rQ   c          	   c   s   t dV }t dV }|dV  |dV  |dV  |jddV }|dddgksXt|dV  |dV  |dV  |jddV }|ddgkstttj" tt	dd|jddV }W d Q R X d S )	Nr   r   r@      T)Zbatchg?)Zseconds)
r	   r   r   r   r   r    r   r!   Zwith_timeoutr   )r"   r#   r$   r%   r   r&   r-   r*   r*   r+   test_get_many   s    

rS   c          	   c   s  |  dV }tdV }||V  t|jddV }td|dV }| V }|jdksXt| t	dd}t
|V  ||V  | V }	|	jd	ksttt |	V  W d Q R X t }
xXy
|	V  W q tk
r   P Y q tk
r
   t |
d
 k sttdV  Y qX qW | V  d S )NrO   r6   T)r5   )r   Zfinishedr   r   errorr   g?)rI   r	   r   r   r7   r   Zstatusr   r   r   r   r   r    	Exceptionr   ZeroDivisionErrorr   r   r8   )r"   r#   r$   r%   r   r6   r9   r:   r'   r(   r)   r*   r*   r+   $test_Future_knows_status_immediately   s0    




rW   c          	   c   sr   |  tdd}t }||V  tdV  | V }tt	 |
 V  W d Q R X | V }t|t	sntd S )Nr   r   g?)r   r   r	   r   r   r   r   r   r    rV   r3   Z	exception
isinstancer   )r"   r#   r$   r%   r'   r6   r(   excr*   r*   r+   test_erred_future   s    

rZ   c             c   sb   t  }x"|j|jd jkr(tdV  qW |  |  x"|j|jd jkr\tdV  q<W d S )Nqueuesg{Gz?)r	   name
extensionsr[   r   r   r8   )r"   r#   r$   r%   r6   r*   r*   r+   
test_close   s    r^   c          	   c   s   t ddd}t }ttj |jddV  W d Q R X t }d||   k rVdk s\n t|dV  t }ttj |jdddV  W d Q R X t }d	||   k rdk sn td S )
Nvr   )maxsizeg333333?)r   g?g       @r@   g?)	r	   r   r   r    r   r!   r   r   r   )r"   r#   r$   r%   r6   r)   stopr*   r*   r+   test_timeout   s    rb   c             #   s@   t    fdd} fdd}| |}| |}||gV  d S )Nc                  s     d d S )N7   )r   r*   )r6   r*   r+   r     s    ztest_2220.<locals>.putc                  s   t    d S )N)printr   r*   )r6   r*   r+   r     s    ztest_2220.<locals>.get)r	   r   )r"   r#   r$   r%   r   r   ZfutZresr*   )r6   r+   	test_2220  s    

re   )-Z
__future__r   r   r   Zdatetimer   r   r   sysr   Ztornador   Zdistributedr   r	   r
   r   r   Zdistributed.metricsZdistributed.utils_testr   r   r   r   r   r   r   r,   r.   r4   r;   Zmarkskipr>   r?   Zskipifversion_inforN   rQ   rS   rW   rZ   r^   rb   re   r*   r*   r*   r+   <module>   s6   !