← Back
Editing: 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
Save File
Cancel