B
    F.\"                 @   sL  d dl mZmZmZ d dl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 d dlZd dlmZmZ d d	lmZ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m Z m!Z!m"Z"m#Z#m$Z$m%Z% d dlm&Z&m'Z'm(Z( d dl)mZ d dl*m+Z+ e'Z,e(Z-edddd Z.ej/0de!dfe dfgdd Z1ej/j2ej34dddeddgd ddd d! Z5d"d# Z6eg ddd$d%d& Z7edgd' dd(d$d)d* Z8d+d, Z9ej/j:e#eddgd- ddd.d/ Z;ej/j<eddgd' d(dd0d1 Z=e#edgd' dd2d$d3d4 Z>ej/j?d5dedd6dgd7d8d9 Z@dS ):    )print_functiondivisionabsolute_import)CancelledError)	timedelta)addN)sleep)delayed)concatsliding_window)ClientwaitNanny)config)time)All)gen_clusterclusterincslowincslowaddslowslowsumbump_rlimit)loopnodebug_setup_modulenodebug_teardown_module)r   )genT)clientc             #   sz   d}  tt|x<tdkrPtdV   fddtdtdD qW d V }|tt tt|ksvtd S )N@      g?c          	      s&   g | ]}  t| |d   qS )r    )submitr   ).0i)cseq <lib/python3.7/site-packages/distributed/tests/test_stress.py
<listcomp>%   s   z!test_stress_1.<locals>.<listcomp>r      )mapr   rangelenr   r   sumAssertionError)r$   sabnresultr&   )r$   r%   r'   test_stress_1   s    
r4   )funcr2   d   i  c       	   
   C   sx   t  h\}\}}t|d | dD}||d}xt|D ]}|||}q8W | |d ks`tW d Q R X W d Q R X d S )Naddress)r   r    r)   )r   r   r!   r+   r3   r.   )	r   r5   r2   r/   r0   r1   r$   xr#   r&   r&   r'   test_stress_gc+   s    r9   winz#test can leave dangling RPC objects)reason)z	127.0.0.1r       )r   ncorestimeoutc       	      g   s   t d}|jjddd}| |}t|gV  |jdd|jdd d  }t|jt|j }xVt	dD ]J}| 
|}x.t|jt d d	 | krtd
V  q~W | |V  qnW d S )Nz
dask.array)2   r?   )r)   r)   )chunksr   )axisr       g      ?g{Gz?)pytestimportorskiprandompersistr   r-   stdr,   daskr+   computeZwaitingr   r   Z_cancel)	r$   r/   workersdar8   yZn_todor#   fr&   r&   r'   test_cancel_stress6   s    

 
 rN   c       
   
   C   s   t d}|jjddd}tdd\}\}}t|d | dj}||}|jd	d
|jdd
 d  }t| x0t	dD ]$}|
|}	tt  ||	 q|W W d Q R X W d Q R X d S )Nz
dask.array)r?   r?   )r)   r)   )r@   
   )Zactive_rpc_timeoutr7   )r   r   )rA   r    rB   )rC   rD   rE   r   r   rF   r-   rG   r   r+   rI   r   Zcancel)
r   rK   r8   r/   r0   r1   r$   rL   r#   rM   r&   r&   r'   test_cancel_stress_syncG   s    

 
rP   )r=   r   r>   c             #   s   d_ td}|jjddd}|d j|d  |jdd }| |}tjfd	d
 t	t
ddt fddtdD V  d S )Ni z
dask.array)i  i  )r6   r6   )sizer@   r    r)   )rA   c             3   sX   t  }xLt  |d k rRt jd jd}|d t| V  | V  td qW d S )NrB   r)   )r=   r   r   zKilled nanny)	r   r   r7   r   startr   r   _closeprint)delayrR   r2   )r/   r&   r'   create_and_destroy_worker`   s    

zDtest_stress_creation_and_deletion.<locals>.create_and_destroy_worker)Zminutesc                s   g | ]} d | qS )g?r&   )r"   r#   )rV   r&   r'   r(   m   s    z5test_stress_creation_and_deletion.<locals>.<listcomp>   )allowed_failuresrC   rD   rE   TZmeanrF   r   	coroutinewith_timeoutr   r   r+   )r$   r/   rK   r8   rL   zr&   )rV   r/   r'   !test_stress_creation_and_deletionU   s    

r]   rO   <   c             '   s  dd l d|_td| fddtt|D V  | j ddV   fddtd	D }d
d tt	d|D }| 
|}d   }t|}ddlm} xtdD ]}tdV  y|  W nN tk
r }  z.||  tdrdd l}|  n W d d } ~ X Y nX |}	|	 V  ||	 qW zyttdd| |V  W n tjk
r   dd |D }
t|j t|
 t| ydd |
  D d }W n tk
r   Y nX tdrdd l}|  n Y n t!k
r   Y nX W d d }X d S )Nr   i  numpyc                s   g | ]} j  d qS )i'  )rE   )r"   r#   )npr&   r'   r(   v   s    z-test_stress_scatter_death.<locals>.<listcomp>r)   )r2   c                s4   g | ],}t td d  dd| dqS )T)pureg?zslowadd-1-%d)rU   dask_key_name)r	   r   choice)r"   r#   )LrE   r&   r'   r(   y   s   r?   c             S   s0   g | ](\}\}}t td d||dd| dqS )T)ra   g{Gz?zslowadd-2-%d)rU   rb   )r	   r   )r"   r#   r0   r1   r&   r&   r'   r(      s   )logger   g?z
log-on-err   )Zsecondsc             S   s   i | ]}|j d kr||jqS )closed)statusr7   )r"   wr&   r&   r'   
<dictcomp>   s    z-test_stress_scatter_death.<locals>.<dictcomp>c             S   s   g | ]}|j r|qS r&   )Zwaiting_for_data)r"   rj   r&   r&   r'   r(      s    )"rE   rX   rC   rD   scatterr+   r,   Z
_replicate	enumerater   rI   listZdistributed.schedulerre   r   r   Zvalidate_state	ExceptionZ	exceptionr   getpdbZ	set_tracerc   rS   remover[   r   Z_gatherTimeoutErrorrT   
processingvaluesr   )r$   r/   rJ   ZaddsZfuturesalivere   r#   rq   rj   Zwsworkerr&   )rd   r`   rE   r'   test_stress_scatter_deathq   sX    
"







rx   c              G   s   t | S )N)r-   )argsr&   r&   r'   vsum   s    rz   P   c       
      '   s   d|_ td td}t|jd d} fddt|D }dd |D } jtd	ftt	|d	g| d
di}| 
| }|V }	t|	tstd S )NFz
dask.arrayresourcei    rW   c                s   g | ]} j j d ddqS ))r6   r6   )rB   rB   )r@   )rE   )r"   r#   )rK   r&   r'   r(      s    z-test_stress_communication.<locals>.<listcomp>c             S   s   g | ]}||j  qS r&   )rY   )r"   r8   r&   r&   r'   r(      s    ZijZdtypeZfloat64)validaterC   rD   r   ZRLIMIT_NOFILEr+   Zatoprz   r
   ziprI   r-   
isinstancefloatr.   )
r$   r/   rJ   r|   r2   ZxsZysr\   futurer3   r&   )rK   r'   test_stress_communication   s    

(r   c             g   s   d|_ x|D ]
}d|_ qW tt}dd tdD }x$tdD ]}dd td|D }q>W tt|}| |}xb|jdkrt	dV  x>td	D ]2}t
|}	t
|}
|	|
k	r||	j|
jd
 qW |jsrP qrW d S )NFc             S   s   g | ]}t t|d dqS )g{Gzt?)rU   )r	   r   )r"   r#   r&   r&   r'   r(      s    z%test_stress_steal.<locals>.<listcomp>r6   rB   c             S   s   g | ]}t t|d dqS )g{Gzt?)rU   )r	   r   )r"   partr&   r&   r'   r(      s   Zfinishedg?   g      ?)r}   r	   r   r+   r   r-   rI   ri   r   r   rE   rc   Z
work_stealr7   rt   )r$   r/   rJ   rj   Zdincrd   r#   Ztotalr   r0   r1   r&   r&   r'   test_stress_steal   s&    




r   x   c       	      g   s   t d}|jjddd}x$tdD ]}|d}|d}q$W | | }xDt|j	 rt
dV  tt|}x|jD ]}|  qW qPW t|V  d S )Nz
dask.array)i  i  )i  r    )rQ   r@   r   )r    i  g      ?)rC   rD   rE   r+   ZrechunkrI   r-   anyrt   ru   r   r   rc   rn   Z_commsabortr   )	r$   r/   rJ   rK   r8   r#   r   rw   Zcommr&   r&   r'   test_close_connections   s    

r   zeIOStream._handle_write blocks on large write_buffer https://github.com/tornadoweb/tornado/issues/2110rW   )r   r>   r=   c          	   c   sL  t d t d}|jd}|j}ddlm} ddlm} ddlm	} ddl
m}	 x,||gD ] }
||||
jd	|
_t |
_q`W |	d
d$}| j|dddV }tdV  W d Q R X |  d }x,||gD ] }
|
jd jd  dk stqW |dd |jD }||d 8 }| |d d k s.t|d |d d k sHtd S )NZcrickr_   i r   )Digest)defaultdict)partial)ResourceProfiler)r   g{Gz?)ZdtTF)Zdirecthashg      ?ztick-durationc             S   s   g | ]
}|j qS r&   )Zmem)r"   tr&   r&   r'   r(     s    z7test_no_delay_during_large_transfer.<locals>.<listcomp>r)   g    .Ag333333?)rC   rD   rE   nbytesZdistributed.counterr   collectionsr   	functoolsr   Zdask.diagnosticsr   Zio_loopZdigestsr   Z
_last_tickrl   r   r   closeZ
componentsmaxr.   ZarrayZresults)r$   r/   rj   r`   r8   Zx_nbytesr   r   r   r   ZserverZrprofr   r   r&   r&   r'   #test_no_delay_during_large_transfer   s,    

 r   )AZ
__future__r   r   r   Zconcurrent.futuresr   Zdatetimer   operatorr   rE   sysr   r   rH   r	   rC   Ztoolzr
   r   Zdistributedr   r   r   Zdistributed.configr   Zdistributed.metricsZdistributed.utilsr   Zdistributed.utils_testr   r   r   r   r   r   r   r   r   r   r   Zdistributed.clientZtornador   Zsetup_moduleZteardown_moduler4   ZmarkZparametrizer9   Zskipifplatform
startswithrN   rP   r]   rx   rz   Zavoid_travisr   skipr   r   Zxfailr   r&   r&   r&   r'   <module>   sJ   ("> 