File: //lib/python3/dist-packages/twisted/test/__pycache__/test_threadpool.cpython-38.pyc
U
W[�V � @ s� d Z ddlmZmZ ddlZddlZddlZddlZddlZddl m
Z
ddl m Z ddl
mZmZmZmZ ddlmZmZ G dd � d e�Ze�e� G d
d � d e j�ZG d d
� d
e j�ZG dd� dej�Z G dd� de�Z G dd� de j�Z dS )z(
Tests for L{twisted.python.threadpool}
� )�division�absolute_importN)�range)�unittest)�
threadpool�
threadable�failure�context)�Team�createMemoryWorkerc @ s& e Zd ZdZdd� Zdd� ZdgZdS )�Synchronizationr c C s || _ || _t�� | _g | _d S �N)�N�waiting� threading�Lock�lock�runs)�selfr r � r �>/usr/lib/python3/dist-packages/twisted/test/test_threadpool.py�__init__ s
zSynchronization.__init__c C s| | j �d�r0t| j�d s$t�d� | j �� n| jd7 _| j �� | j�d � t| j�| j krn| j
�� | j �� d S )NF� g-C ��6*?� ) r �acquire�lenr �time�sleep�release�failures�appendr r �r r r r �run! s
zSynchronization.runr" N)�__name__�
__module__� __qualname__r r r" Z synchronizedr r r r r s r c @ s� e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
d � Zd d
� Z dd� Z
dd� Z dd� Z dd� Z
dd� Zdd� Zdd� Zd d � Zd d � Zd d!� Zd"d#� Zd$d%� Zd&S )'�ThreadPoolTestsz
Test threadpools.
c C s dS )zD
Return number of seconds to wait before giving up.
r r r! r r r �
getTimeoutF s zThreadPoolTests.getTimeoutc C s8 t d�}|D ] }|�d�r q4t�d� q | �d� d S )Ni@B Fg�h㈵��>z%A long time passed without succeeding)r r r r �fail)r r �items�ir r r � _waitForLockM s
z ThreadPoolTests._waitForLockc C s, t �dd�}| �|jd� | �|jd� dS )zy
L{ThreadPool.min} and L{ThreadPool.max} are set to the values passed to
L{ThreadPool.__init__}.
� � N)r �
ThreadPool� assertEqual�min�max�r �poolr r r �test_attributesW s z ThreadPoolTests.test_attributesc C sz t �dd�}|�� | �|j� | �t|j�d� t �dd�}| �t|j�d� |�� | �|j� | �t|j�d� dS )zV
L{ThreadPool.start} creates the minimum number of threads specified.
r r � �
N)r r. �start�
addCleanup�stopr/ r �threadsr2 r r r �
test_starta s zThreadPoolTests.test_startc C s< t �dd�}|�� |�� |�d� | �t|j�d� dS )z�
L{ThreadPool.adjustPoolsize} only modifies the pool size and does not
start new workers while the pool is not running.
r r � N)r r. r7 r9 ZadjustPoolsizer/ r r: r2 r r r � test_adjustingWhenPoolStoppedq s
z-ThreadPoolTests.test_adjustingWhenPoolStoppedc C s� t �dd�}|�� | �|j� | �|jg � dd� }G dd� dt�}|� }t� |�}t� |�}|�
||� t � � }|�
|j
� |�| �� � ~~t�� | �|� � | �|� � dS )z�
Test that creating threads in the threadpool with application-level
objects as arguments doesn't results in those objects never being
freed, with the thread maintaining a reference to them as long as it
exists.
r r c S s d S r
r )�argr r r �worker� s z<ThreadPoolTests.test_threadCreationArguments.<locals>.workerc @ s e Zd ZdS )z:ThreadPoolTests.test_threadCreationArguments.<locals>.DumbN�r# r$ r% r r r r �Dumb� s rA N)r r. r7 r8 r9 r/ r: �object�weakref�ref� callInThreadr �Event�set�waitr' �gc�collect� assertIsNone)r �tpr? rA �unique� workerRef� uniqueRef�eventr r r � test_threadCreationArguments} s$
z,ThreadPoolTests.test_threadCreationArgumentsc s, t �dd�}|�� ��|j� ��|jg � i �t�� �t�� �g ��������fdd�}� fdd�}G dd� dt �� � � }t
� |�}t
� |��t
� |��|j ||||d � ~~��
� ����� � t�� ���� � ���� � ~t�� ��|� � ���d � � ��t��� �d
d
g� d
S ) ze
As C{test_threadCreationArguments} above, but for
callInThreadWithCallback.
r r c sF t �� ����� � �� �d< �� �d< � �� ��t�|�� d S )NrN rO )rI rJ rH r' rG r rC rD ��success�result)� onResultDone� onResultWait�refdict� resultRefr rO rN r r �onResult� s
zVThreadPoolTests.test_threadCreationArgumentsCallInThreadWithCallback.<locals>.onResultc s � � S r
r )r> �test)rA r r r? � s zTThreadPoolTests.test_threadCreationArgumentsCallInThreadWithCallback.<locals>.workerc @ s e Zd ZdS )zRThreadPoolTests.test_threadCreationArgumentsCallInThreadWithCallback.<locals>.DumbNr@ r r r r rA � s rA )rZ N)r r. r7 r8 r9 r/ r: r rF rB rC rD �callInThreadWithCallbackrG rH r' rI rJ rK �list�values)r rL rY r? rM Z onResultRefr )rA rU rV rW rX r rO rN r �4test_threadCreationArgumentsCallInThreadWithCallback� s8
zDThreadPoolTests.test_threadCreationArgumentsCallInThreadWithCallbackc C sX t �dd�}| �|jd� | �|jd� t�t�|��}| �|jd� | �|jd� dS )z�
Threadpools can be pickled and unpickled, which should preserve the
number of threads and other parameters.
� � N)r r. r/ r0 r1 �pickle�loads�dumps)r r3 �copyr r r �test_persistence� s z ThreadPoolTests.test_persistencec C sv d}t �� }|�� | �|j� t�� }|�� t||�}t |�D ]}|||� qB| �
|� | � |j d�
|j �� dS )z�
Test synchronization of calls made with C{method}, which should be
one of the mechanisms of the threadpool to execute work in threads.
r6 zrun() re-entered {} timesN)r r. r7 r8 r9 r r r r r r+ � assertFalser �format)r �methodr rL r �actorr* r r r �_threadpoolTest s
�z ThreadPoolTests._threadpoolTestc C s | � dd� �S )z?
Call C{_threadpoolTest} with C{callInThread}.
c S s | � |j�S r
)rE r" )rL ri r r r �<lambda>& � z3ThreadPoolTests.test_callInThread.<locals>.<lambda>)rj r! r r r �test_callInThread! s �z!ThreadPoolTests.test_callInThreadc s` G dd� dt �� � fdd�}t�dd�}|�|� |�� |�� | �� �}| �t|�d� dS )zi
L{ThreadPool.callInThread} logs exceptions raised by the callable it
is passed.
c @ s e Zd ZdS )z<ThreadPoolTests.test_callInThreadException.<locals>.NewErrorNr@ r r r r �NewError. s rn c s
� � �d S r
r r �rn r r �
raiseError1 s z>ThreadPoolTests.test_callInThreadException.<locals>.raiseErrorr r N) � Exceptionr r. rE r7 r9 �flushLoggedErrorsr/ r )r rp rL �errorsr ro r �test_callInThreadException) s
z*ThreadPoolTests.test_callInThreadExceptionc s� t �� ���� g � � �fdd�}t�dd�}|�|dd� � |�� z| ��� W 5 |�� X | � � d � | �
� d d� dS ) z�
L{ThreadPool.callInThreadWithCallback} calls C{onResult} with a
two-tuple of C{(True, result)} where C{result} is the value returned
by the callable supplied.
c s �� � � �| � � �|� d S r
�r r rR ��results�waiterr r rY H s
z?ThreadPoolTests.test_callInThreadWithCallback.<locals>.onResultr r c S s dS )NrZ r r r r r rk N rl z?ThreadPoolTests.test_callInThreadWithCallback.<locals>.<lambda>rZ N) r r r r r. r[ r7 r9 r+ �
assertTruer/ )r rY rL r rv r � test_callInThreadWithCallback= s
z-ThreadPoolTests.test_callInThreadWithCallbackc s� G dd� dt �� � fdd�}t�� ���� g ���fdd�}t�dd�}|�||� |�� z| � �� W 5 |�� X | �
�d � | � �d t j
� | �t�d j� �� d S )
z�
L{ThreadPool.callInThreadWithCallback} calls C{onResult} with a
two-tuple of C{(False, failure)} where C{failure} represents the
exception raised by the callable supplied.
c @ s e Zd ZdS )zRThreadPoolTests.test_callInThreadWithCallbackExceptionInCallback.<locals>.NewErrorNr@ r r r r rn ` s rn c s
� � �d S r
r r ro r r rp c s zTThreadPoolTests.test_callInThreadWithCallbackExceptionInCallback.<locals>.raiseErrorc s �� � � �| � � �|� d S r
ru rR rv r r rY k s
zRThreadPoolTests.test_callInThreadWithCallbackExceptionInCallback.<locals>.onResultr r N)rq r r r r r. r[ r7 r9 r+ rf ZassertIsInstancer ZFailurery �
issubclass�type)r rp rY rL r )rn rw rx r �0test_callInThreadWithCallbackExceptionInCallbackZ s
z@ThreadPoolTests.test_callInThreadWithCallbackExceptionInCallbackc s� G dd� dt �� t�� }|�� g �� �fdd�}t�dd�}|�|dd� � |�|j� |� � z| � |� W 5 |�
� X | � � �}| �
t|�d� | ��d � | ��d � d S )
zj
L{ThreadPool.callInThreadWithCallback} logs the exception raised by
C{onResult}.
c @ s e Zd ZdS )zRThreadPoolTests.test_callInThreadWithCallbackExceptionInOnResult.<locals>.NewErrorNr@ r r r r rn � s rn c s �� | � �� |� � � �d S r
)r rR �rn rw r r rY � s
zRThreadPoolTests.test_callInThreadWithCallbackExceptionInOnResult.<locals>.onResultr r c S s d S r
r r r r r rk � rl zRThreadPoolTests.test_callInThreadWithCallbackExceptionInOnResult.<locals>.<lambda>N)rq r r r r r. r[ rE r r7 r9 r+ rr r/ r ry rK )r rx rY rL rs r r~ r �0test_callInThreadWithCallbackExceptionInOnResult~ s
z@ThreadPoolTests.test_callInThreadWithCallbackExceptionInOnResultc s� g �t �� � � �fdd�}�fdd�}t�dd�}|�||� |�� | �|j� � �| � � � | �
t ��d� | �
�d �d � dS ) z�
L{ThreadPool.callInThreadWithCallback} calls the function it is
given and the C{onResult} callback in the same thread.
c s �� t�� j� � �� d S r
)r r �
currentThread�identrG rR �rP � threadIdsr r rY � s z5ThreadPoolTests.test_callbackThread.<locals>.onResultc s � � t�� j� d S r
)r r r� r� r )r� r r �func� s z1ThreadPoolTests.test_callbackThread.<locals>.funcr r r< N) r rF r r. r[ r7 r8 r9 rH r' r/ r )r rY r� rL r r� r �test_callbackThread� s z#ThreadPoolTests.test_callbackThreadc s� t j�� jd }d|d< g � t�� �� �fdd�}� fdd�}t�dd �}|�||� |� � | �
|j � �� | �
� � | �t� �d
� | �|� d � | �|� d � d S ) z�
The context L{ThreadPool.callInThreadWithCallback} is invoked in is
shared by the context the callable and C{onResult} callback are
invoked in.
���zthis must be presentZtestingc s& t j�� jd }� �|� ��� d S �Nr� )r �theContextTracker�currentContext�contextsr rG )rS rT �ctx�r� rP r r rY � s
z6ThreadPoolTests.test_callbackContext.<locals>.onResultc s t j�� jd } � �| � d S r� )r r� r� r� r )r� )r� r r r� � s z2ThreadPoolTests.test_callbackContext.<locals>.funcr r r< N)r r� r� r� r rF r r. r[ r7 r8 r9 rH r' r/ r )r ZmyctxrY r� rL r r� r �test_callbackContext� s z$ThreadPoolTests.test_callbackContextc C sN t �� }|�� t�dd�}|�|j� |�� z| � |� W 5 |�� X dS )z�
Work added to the threadpool before its start should be executed once
the threadpool is started: this is ensured by trying to release a lock
previously acquired.
r r N)
r r r r r. rE r r7 r9 r+ )r rx rL r r r �test_existingWork� s z!ThreadPoolTests.test_existingWorkc s� t �dd�}|�� | �|j� | �|jd� | �t|j�d� | �t|j �d� t
� � �t
� � � � �fdd�}|� |� ��
d� | �|jd� | �t|j�d� | �t|j �d� � �� t|j�s�t�d� q�| �t|j�d� | �t|j �d� dS )z{
As the worker receives and completes work, it transitions between
the working and waiting states.
r r c s �� � � �d� d S )Nr6 )rG rH r �Z threadFinishZ
threadWorkingr r �_thread s z;ThreadPoolTests.test_workerStateTransition.<locals>._threadr6 g����Mb@?N)r r. r7 r8 r9 r/ �workersr �waitersZworkingr rF rE rH rG r r )r r3 r� r r� r �test_workerStateTransition� s&
z*ThreadPoolTests.test_workerStateTransitionN)r# r$ r% �__doc__r' r+ r4 r; r= rQ r^ re rj rm rt rz r} r r� r� r� r� r r r r r& A s&
+K $#"r& c @ s$ e Zd Zdd� Zdd� Zdd� ZdS )�RaceConditionTestsc s< t �dd�� _ t�� � _� j �� � fdd�}� �|� d S )Nr r6 c s � j �� � ` d S r
)r r9 r r! r r �done s
z&RaceConditionTests.setUp.<locals>.done)r r. r rF rP r7 r8 )r r� r r! r �setUp s
zRaceConditionTests.setUpc C s dS )z=
A reasonable number of seconds to time out.
r r r! r r r r' % s z RaceConditionTests.getTimeoutc C s� | � � }| j�| jj� | j�|� | j�� td�D ]}| j�| jj� q6| j�| jj� | j�|� | j�� s�| j�� | � d� dS )a�
If multiple threads are waiting on an event (via blocking on something
in a callable passed to L{threadpool.ThreadPool.callInThread}), and
there is spare capacity in the threadpool, sending another callable
which will cause those to un-block to
L{threadpool.ThreadPool.callInThread} will reliably run that callable
and un-block the blocked threads promptly.
@note: This is not really a unit test, it is a stress-test. You may
need to run it with C{trial -u} to fail reliably if there is a
problem. It is very hard to regression-test for this particular
bug - one where the thread pool may consider itself as having
"enough capacity" when it really needs to spin up a new thread if
it possibly can - in a deterministic way, since the bug can only be
provoked by subtle race conditions.
r5 z9'set' did not run in thread; timed out waiting on 'wait'.N)
r' r rE rP rG rH �clearr ZisSetr( )r Ztimeoutr* r r r �test_synchronization, s
�z'RaceConditionTests.test_synchronizationN)r# r$ r% r� r' r� r r r r r� s r� c @ s e Zd ZdZdd� Zdd� ZdS )�
MemoryPoolz
A deterministic threadpool that uses in-memory data structures to queue
work rather than threads to execute work.
c O s* || _ || _|| _tjj| f|�|� dS )a�
Initialize this L{MemoryPool} with a test case.
@param coordinator: a worker used to coordinate work in the L{Team}
underlying this threadpool.
@type coordinator: L{twisted._threads.IExclusiveWorker}
@param failTest: A 1-argument callable taking an exception and raising
a test-failure exception.
@type failTest: 1-argument callable taking (L{Failure}) and raising
L{unittest.FailTest}.
@param newWorker: a 0-argument callable that produces a new
L{twisted._threads.IWorker} provider on each invocation.
@type newWorker: 0-argument callable returning
L{twisted._threads.IWorker}.
N)� _coordinator� _failTest�
_newWorkerr r. r )r � coordinatorZfailTest� newWorker�args�kwargsr r r r S s zMemoryPool.__init__c s& � ��fdd�}t �j|�jd���S )a�
Override testing hook to create a deterministic threadpool.
@param currentLimit: A 1-argument callable which returns the current
threadpool size limit.
@param threadFactory: ignored in this invocation; a 0-argument callable
that would produce a thread.
@return: a L{Team} backed by the coordinator and worker passed to
L{MemoryPool.__init__}.
c s& �� � } | j| j � � kr d S ��� S r
)Z
statisticsZbusyWorkerCountZidleWorkerCountr� )Zstats�� currentLimitr Zteamr r � respectLimitx s ��z&MemoryPool._pool.<locals>.respectLimit)r� Z createWorkerZ logException)r
r� r� )r r� Z
threadFactoryr� r r� r �_poolk s
�zMemoryPool._poolN)r# r$ r% r� r r� r r r r r� M s r� c @ s e Zd ZdZdd� Zdd� ZdS )�
PoolHelpera
A L{PoolHelper} constructs a L{threadpool.ThreadPool} that doesn't actually
use threads, by using the internal interfaces in L{twisted._threads}.
@ivar performCoordination: a 0-argument callable that will perform one unit
of "coordination" - work involved in delegating work to other threads -
and return L{True} if it did any work, L{False} otherwise.
@ivar workers: the workers which represent the threads within the pool -
the workers other than the coordinator.
@type workers: L{list} of 2-tuple of (L{IWorker}, C{workPerformer}) where
C{workPerformer} is a 0-argument callable like C{performCoordination}.
@ivar threadpool: a modified L{threadpool.ThreadPool} to test.
@type threadpool: L{MemoryPool}
c s: t � \}� _g � _� fdd�}t||j|f|�|�� _dS )z�
Create a L{PoolHelper}.
@param testCase: a test case attached to this helper.
@type args: The arguments passed to a L{threadpool.ThreadPool}.
@type kwargs: The arguments passed to a L{threadpool.ThreadPool}
c s � j �t� � � j d d S )Nr� r )r� r r r r! r r r� � s z&PoolHelper.__init__.<locals>.newWorkerN)r �performCoordinationr� r� r( r )r ZtestCaser� r� r� r� r r! r r � s
��zPoolHelper.__init__c C s | � � r
q dS )z�
Perform all currently scheduled "coordination", which is the work
involved in delegating work to other threads.
N)r� r! r r r �performAllCoordination� s z!PoolHelper.performAllCoordinationN)r# r$ r% r� r r� r r r r r� � s r� c @ s e Zd ZdZdd� Zdd� ZdS )�MemoryBackedTestszn
Tests using L{PoolHelper} to deterministically test properties of the
threadpool implementation.
c C sl t | dd�}d}t|�D ]}|j�dd� � q|�� | �|jg � |j�� |�� | �t|j�|� dS )z�
If a threadpool is told to do work before starting, then upon starting
up, it will start enough workers to handle all of the enqueued work
that it's been given.
r r6 r c S s d S r
r r r r r rk � rl z;MemoryBackedTests.test_workBeforeStarting.<locals>.<lambda>N) r� r r rE r� r/ r� r7 r �r �helper�n�xr r r �test_workBeforeStarting� s
z)MemoryBackedTests.test_workBeforeStartingc C sp t | dd�}d}t|�D ]}|j�dd� � q|�� | �|jg � |j�� |�� | �t|j�|jj � dS )z�
If the amount of work before starting exceeds the maximum number of
threads allowed to the threadpool, only the maximum count will be
started.
r r6 �2 c S s d S r
r r r r r rk � rl zBMemoryBackedTests.test_tooMuchWorkBeforeStarting.<locals>.<lambda>N)
r� r r rE r� r/ r� r7 r r1 r� r r r � test_tooMuchWorkBeforeStarting� s
z0MemoryBackedTests.test_tooMuchWorkBeforeStartingN)r# r$ r% r� r� r� r r r r r� � s r� ) r� Z
__future__r r ra r rC rI r Ztwisted.python.compatr Z
twisted.trialr Ztwisted.pythonr r r r Ztwisted._threadsr
r rB r Z synchronizeZSynchronousTestCaser& r� r. r� r� r� r r r r �<module> s( %
Y6;1