File: //lib/python3/dist-packages/twisted/test/__pycache__/test_threadpool.cpython-310.pyc
o
�b�U � @ s� d Z ddlZddlZddlZddlZddlZddlmZmZ ddl m
Z
m Z m Z m
Z
ddlmZ G dd� d�Ze �e� G dd � d ej�ZG d
d � d ej�ZG d d
� d
e
j�ZG dd� d�ZG dd� dej�ZdS )z(
Tests for L{twisted.python.threadpool}
� N)�Team�createMemoryWorker)�context�failure�
threadable�
threadpool)�unittestc @ 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�rt| j�d st�d� | j �� n| jd7 _| j �� | j�d � t| j�| j kr7| 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 � 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 � dS )zD
Return number of seconds to wait before giving up.
r r r r r r �
getTimeout@ � zThreadPoolTests.getTimeoutc C s: t d�}|D ]}|�d�r d S t�d� q| �d� d S )Ni@B Fg�h㈵��>z%A long time passed without succeeding)�ranger r r �fail)r r �items�ir r r � _waitForLockF 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_attributesO 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�stopr0 r �threadsr3 r r r �
test_startX 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/ r8 r: �adjustPoolsizer0 r r; r3 r r r � test_adjustingWhenPoolStoppedg s
z-ThreadPoolTests.test_adjustingWhenPoolStoppedc C s� t �dd�}|�� | �|j� | �|jg � dd� }G dd� d�}|� }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 � d S r
r )�argr r r �worker� s z<ThreadPoolTests.test_threadCreationArguments.<locals>.workerc @ � e Zd ZdS )z:ThreadPoolTests.test_threadCreationArguments.<locals>.DumbN�r r! r" r r r r �Dumb� � rE N)r r/ r8 r9 r: r0 r; �weakref�ref� callInThreadr
�Event�set�waitr& �gc�collect� assertIsNone)r �tprB rE �unique� workerRef� uniqueRef�eventr r r � test_threadCreationArgumentsr s$
z,ThreadPoolTests.test_threadCreationArgumentsc s* t �dd�}|�� ��|j� ��|jg � i �t�� �t�� �g ��������fdd�}� fdd�}G dd� d�� � � }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 )NrR rS )rM rN rL r& rK r rG rH ��success�result)� onResultDone� onResultWait�refdict� resultRefr rS rR r r �onResult� s
zVThreadPoolTests.test_threadCreationArgumentsCallInThreadWithCallback.<locals>.onResultc s � � S r
r )rA �test)rE r r rB � � zTThreadPoolTests.test_threadCreationArgumentsCallInThreadWithCallback.<locals>.workerc @ rC )zRThreadPoolTests.test_threadCreationArgumentsCallInThreadWithCallback.<locals>.DumbNrD r r r r rE � rF rE )r^ N)r r/ r8 r9 r: r0 r; r
rJ rG rH �callInThreadWithCallbackrK rL r& rM rN rO �list�values)r rP r] rB rQ � onResultRefr )rE rY rZ r[ r\ r rS rR 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/ r0 r1 r2 �pickle�loads�dumps)r r4 �copyr r r �test_persistence� s z ThreadPoolTests.test_persistencec C sx d}t �� }|�� | �|j� t�� }|�� t||�}t |�D ]}|||� q!| �
|� | � |j d|j � d�� 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.
r7 zrun() re-entered z timesN)
r r/ r8 r9 r: r
r r r r( r, � assertFalser )r �methodr rP 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
)rI r )rP rn r r r �<lambda> s z3ThreadPoolTests.test_callInThread.<locals>.<lambda>)ro 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 @ rC )z<ThreadPoolTests.test_callInThreadException.<locals>.NewErrorNrD r r r r �NewError rF rr c � � � �r
r r �rr r r �
raiseError r_ z>ThreadPoolTests.test_callInThreadException.<locals>.raiseErrorr r N) � Exceptionr r/ rI r8 r: �flushLoggedErrorsr0 r )r ru rP �errorsr rt r �test_callInThreadException s
z*ThreadPoolTests.test_callInThreadExceptionc s� t �� ���� g � � �fdd�}t�dd�}|�|dd� � |�� z | ��� W |�� n|�� w | � � 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 � �� � � �| � � �|� d S r
�r r rV ��results�waiterr r r] 5 �
z?ThreadPoolTests.test_callInThreadWithCallback.<locals>.onResultr r c S r% )Nr^ r r r r r rp ; � z?ThreadPoolTests.test_callInThreadWithCallback.<locals>.<lambda>r^ N) r
r r r r/ r` r8 r, r: �
assertTruer0 )r r] rP r r| r � test_callInThreadWithCallback* s z-ThreadPoolTests.test_callInThreadWithCallbackc s� G dd� dt �� � fdd�}t�� ���� g ���fdd�}t�dd�}|�||� |�� z | ��� W |� � n|� � w | �
�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 @ rC )zRThreadPoolTests.test_callInThreadWithCallbackExceptionInCallback.<locals>.NewErrorNrD r r r r rr M rF rr c rs r
r r rt r r ru P r_ zTThreadPoolTests.test_callInThreadWithCallbackExceptionInCallback.<locals>.raiseErrorc rz r
r{ rV r| r r r] X r zRThreadPoolTests.test_callInThreadWithCallbackExceptionInCallback.<locals>.onResultr r N)rv r
r r r r/ r` r8 r, r: rl �assertIsInstancer �Failurer� �
issubclass�type)r ru r] rP r )rr r} r~ r �0test_callInThreadWithCallbackExceptionInCallbackF s z@ThreadPoolTests.test_callInThreadWithCallbackExceptionInCallbackc s� G dd� dt �� t�� }|�� g �� �fdd�}t�dd�}|�|dd� � |�|j� |� � z | �
|� W |� � n|� � w | � � �}| �
t|�d� | ��d � | ��d � d S )
zj
L{ThreadPool.callInThreadWithCallback} logs the exception raised by
C{onResult}.
c @ rC )zRThreadPoolTests.test_callInThreadWithCallbackExceptionInOnResult.<locals>.NewErrorNrD r r r r rr p rF rr c s �� | � �� |� � � �r
)r rV �rr r} r r r] x s
zRThreadPoolTests.test_callInThreadWithCallbackExceptionInOnResult.<locals>.onResultr r c S r@ r
r r r r r rp ~ r� zRThreadPoolTests.test_callInThreadWithCallbackExceptionInOnResult.<locals>.<lambda>N)rv r
r r r r/ r` rI r r8 r, r: rw r0 r r� rO )r r~ r] rP rx r r� r �0test_callInThreadWithCallbackExceptionInOnResultj 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
�current_thread�identrK rV �rT � threadIdsr r r] � 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
rJ r r/ r` r8 r9 r: rL r& r0 r )r r] r� rP 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 present�testingc s& t j�� jd }� �|� ��� d S �Nr� )r �theContextTracker�currentContext�contextsr rK )rW rX �ctx�r� rT r r r] � 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
rJ r r/ r` r8 r9 r: rL r& r0 r )r �myctxr] r� rP r r� r �test_callbackContext� s z$ThreadPoolTests.test_callbackContextc C sT t �� }|�� t�dd�}|�|j� |�� z | �|� W |� � dS |� � w )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/ rI r r8 r, r: )r r~ rP 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�snt�d� t|j�rd| �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 )Nr7 )rK rL r �� threadFinish�
threadWorkingr r �_thread� s z;ThreadPoolTests.test_workerStateTransition.<locals>._threadr7 g����Mb@?N)r r/ r8 r9 r: r0 �workersr �waiters�workingr
rJ rI rL rK r r )r r4 r� r r� r �test_workerStateTransition� s(
�z*ThreadPoolTests.test_workerStateTransitionN)r r! r" �__doc__r& r, r5 r< r? rU rd rk ro rq ry r� r� r� r� r� r� r� r r r r r$ ; s( *J $#! r$ c @ s$ e Zd Zdd� Zdd� Zdd� ZdS )�RaceConditionTestsc s< t �dd�� _ t�� � _� j �� � fdd�}� �|� d S )Nr r7 c s � j �� � ` d S r
)r r: r r r r �done s
z&RaceConditionTests.setUp.<locals>.done)r r/ r
rJ rT r8 r9 )r r� r r r �setUp s
zRaceConditionTests.setUpc C r% )z=
A reasonable number of seconds to time out.
r r r r r r r& r' z RaceConditionTests.getTimeoutc C s� | � � }| j�| jj� | j�|� | j�� td�D ]
}| j�| jj� q| j�| jj� | j�|� | j�� sE| j�� | � d� dS 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.
r6 z9'set' did not run in thread; timed out waiting on 'wait'.N)
r& r rI rT rK rL �clearr( �isSetr) )r �timeoutr+ r r r �test_synchronization s
�z'RaceConditionTests.test_synchronizationN)r r! r" r� r& r� r r r r r� � s r� c @ � 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 s0 || _ || _|| _tjj| g|�R i |�� 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 � coordinator�failTest� newWorker�args�kwargsr r r r 5 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 � � krd S ��� S r
)�
statistics�busyWorkerCount�idleWorkerCountr� )�stats�� currentLimitr �teamr r � respectLimitZ s z&MemoryPool._pool.<locals>.respectLimit)r� � createWorker� logException)r r� r� )r r� �
threadFactoryr� r r� r �_poolL s �zMemoryPool._poolN)r r! r" r� r r� r r r r r� / s r� c @ r� )�
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|g|�R i |��� _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 �testCaser� r� r� r� r r r r } s
�� �zPoolHelper.__init__c C s | � � r | � � sdS 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� k s r� c @ r� )�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 r7 r c S r@ r
r r r r r rp � r� z;MemoryBackedTests.test_workBeforeStarting.<locals>.<lambda>N) r� r( r rI r� r0 r� r8 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 r7 �2 c S r@ r
r r r r r rp � r� zBMemoryBackedTests.test_tooMuchWorkBeforeStarting.<locals>.<lambda>N)
r� r( r rI r� r0 r� r8 r r2 r� r r r � test_tooMuchWorkBeforeStarting� s
z0MemoryBackedTests.test_tooMuchWorkBeforeStartingN)r r! r" r� r� r� r r r r r� � s r� )r� rM rg r
r rG �twisted._threadsr r �twisted.pythonr r r r �
twisted.trialr r � synchronize�SynchronousTestCaser$ r� r/ r� r� r� r r r r �<module> s&
$ G0<0