File manager - Edit - /home/newsbmcs.com/public_html/static/img/logo/_threads.tar
Back
_pool.py 0000644 00000004334 15030470614 0006232 0 ustar 00 # -*- test-case-name: twisted._threads.test -*- # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Top level thread pool interface, used to implement L{twisted.python.threadpool}. """ from queue import Queue from threading import Lock, Thread, local as LocalStorage from twisted.python.log import err from ._team import Team from ._threadworker import LockWorker, ThreadWorker def pool(currentLimit, threadFactory=Thread): """ Construct a L{Team} that spawns threads as a thread pool, with the given limiting function. @note: Future maintainers: while the public API for the eventual move to twisted.threads should look I{something} like this, and while this function is necessary to implement the API described by L{twisted.python.threadpool}, I am starting to think the idea of a hard upper limit on threadpool size is just bad (turning memory performance issues into correctness issues well before we run into memory pressure), and instead we should build something with reactor integration for slowly releasing idle threads when they're not needed and I{rate} limiting the creation of new threads rather than just hard-capping it. @param currentLimit: a callable that returns the current limit on the number of workers that the returned L{Team} should create; if it already has more workers than that value, no new workers will be created. @type currentLimit: 0-argument callable returning L{int} @param threadFactory: Factory that, when given a C{target} keyword argument, returns a L{threading.Thread} that will run that target. @type threadFactory: callable returning a L{threading.Thread} @return: a new L{Team}. """ def startThread(target): return threadFactory(target=target).start() def limitedWorkerCreator(): stats = team.statistics() if stats.busyWorkerCount + stats.idleWorkerCount >= currentLimit(): return None return ThreadWorker(startThread, Queue()) team = Team( coordinator=LockWorker(Lock(), LocalStorage()), createWorker=limitedWorkerCreator, logException=err, ) return team __pycache__/_threadworker.cpython-310.pyc 0000644 00000007050 15030470614 0014317 0 ustar 00 o �b� � @ s^ d Z ddlmZ ddlmZ ddlmZ e� Zee�G dd� d��Z ee�G dd � d ��Z d S ) zE Implementation of an L{IWorker} based on native threads and queues. � )� implementer� )�Quit)�IExclusiveWorkerc @ �( e Zd ZdZdd� Zdd� Zdd� ZdS ) � ThreadWorkerz� An L{IExclusiveWorker} implemented based on a single thread and a queue. This worker ensures exclusivity (i.e. it is an L{IExclusiveWorker} and not an L{IWorker}) by performing all of the work passed to C{do} on the I{same} thread. c s& � | _ t� | _� fdd�}||� dS )a� Create a L{ThreadWorker} with a function to start a thread and a queue to use to communicate with that thread. @param startThread: a callable that takes a callable to run in another thread. @type startThread: callable taking a 0-argument callable and returning nothing. @param queue: A L{Queue} to use to give tasks to the thread created by C{startThread}. @type queue: L{Queue} c s t � jt�D ]} | � qd S )N)�iter�get�_stop)�task��queue� �@/usr/lib/python3/dist-packages/twisted/_threads/_threadworker.py�work- s �z#ThreadWorker.__init__.<locals>.workN)�_qr �_hasQuit)�self� startThreadr r r r r �__init__ s zThreadWorker.__init__c C s | j �� | j�|� dS )z� Perform the given task on the thread owned by this L{ThreadWorker}. @param task: the function to call on a thread. N)r �checkr �put)r r r r r �do3 � zThreadWorker.doc C s | j �� | j�t� dS )zT Reject all future work and stop the thread started by C{__init__}. N)r �setr r r �r r r r �quit< r zThreadWorker.quitN��__name__� __module__� __qualname__�__doc__r r r r r r r r s r c @ r ) � LockWorkerzE An L{IWorker} implemented based on a mutual-exclusion lock. c C s t � | _|| _|| _dS )z� @param lock: A mutual-exclusion lock, with C{acquire} and C{release} methods. @type lock: L{threading.Lock} @param local: Local storage. @type local: L{threading.local} N)r �_quit�_lock�_local)r �lock�localr r r r L s zLockWorker.__init__c C s� | j }| j}| j�� t|dd�}|du rJg }|_|�|� |�� z |r8|�d�� |s&W |� � d|_dS W |� � d|_dS |� � d|_w |�|� dS )z� Do the given work on this thread, with the mutex acquired. If this is called re-entrantly, return and wait for the outer invocation to do the work. @param work: the work to do with the lock held. �workingNr ) r$ r% r# r �getattrr( �append�acquire�pop�release)r r r& r' r( r r r r Y s( � � �z LockWorker.doc C s | j �� d| _dS )z* Quit this L{LockWorker}. N)r# r r$ r r r r r r s zLockWorker.quitNr r r r r r" F s r" N) r! �zope.interfacer � _conveniencer � _ithreadsr �objectr r r" r r r r �<module> s 3 __pycache__/_team.cpython-310.pyc 0000644 00000016744 15030470614 0012556 0 ustar 00 o �b� � @ s\ d Z ddlmZ ddlmZ ddlmZ ddlmZ G dd� d�Z ee�G d d � d ��Z d S ) zZ Implementation of a L{Team} of workers; a thread-pool that can allocate work to workers. � )�deque)� implementer� )�IWorker)�Quitc @ s e Zd ZdZdd� ZdS )� Statisticsa� Statistics about a L{Team}'s current activity. @ivar idleWorkerCount: The number of idle workers. @type idleWorkerCount: L{int} @ivar busyWorkerCount: The number of busy workers. @type busyWorkerCount: L{int} @ivar backloggedWorkCount: The number of work items passed to L{Team.do} which have not yet been sent to a worker to be performed because not enough workers are available. @type backloggedWorkCount: L{int} c C s || _ || _|| _d S �N)�idleWorkerCount�busyWorkerCount�backloggedWorkCount)�selfr r r � r �8/usr/lib/python3/dist-packages/twisted/_threads/_team.py�__init__# s zStatistics.__init__N)�__name__� __module__� __qualname__�__doc__r r r r r r s r c @ s\ e Zd ZdZdd� Zdd� Zdd� Zdd d �Zdd d �Zd d� Z dd� Z dd� Z dd� Z dS )�Teamax A composite L{IWorker} implementation. @ivar _quit: A L{Quit} flag indicating whether this L{Team} has been quit yet. This may be set by an arbitrary thread since L{Team.quit} may be called from anywhere. @ivar _coordinator: the L{IExclusiveWorker} coordinating access to this L{Team}'s internal resources. @ivar _createWorker: a callable that will create new workers. @ivar _logException: a 0-argument callable called in an exception context when there is an unhandled error from a task passed to L{Team.do} @ivar _idle: a L{set} of idle workers. @ivar _busyCount: the number of workers currently busy. @ivar _pending: a C{deque} of tasks - that is, 0-argument callables passed to L{Team.do} - that are outstanding. @ivar _shouldQuitCoordinator: A flag indicating that the coordinator should be quit at the next available opportunity. Unlike L{Team._quit}, this flag is only set by the coordinator. @ivar _toShrink: the number of workers to shrink this L{Team} by at the next available opportunity; set in the coordinator. c C s@ t � | _|| _|| _|| _t� | _d| _t� | _ d| _ d| _ dS )a� @param coordinator: an L{IExclusiveWorker} which will coordinate access to resources on this L{Team}; that is to say, an L{IExclusiveWorker} whose C{do} method ensures that its given work will be executed in a mutually exclusive context, not in parallel with other work enqueued by C{do} (although possibly in parallel with the caller). @param createWorker: A 0-argument callable that will create an L{IWorker} to perform work. @param logException: A 0-argument callable called in an exception context when the work passed to C{do} raises an exception. r FN) r �_quit� _coordinator� _createWorker� _logException�set�_idle� _busyCountr �_pending�_shouldQuitCoordinator� _toShrink)r � coordinator� createWorker� logExceptionr r r r I s z Team.__init__c C s t t| j�| jt| j��S )z� Gather information on the current status of this L{Team}. @return: a L{Statistics} describing the current state of this L{Team}. )r �lenr r r �r r r r � statisticsd s zTeam.statisticsc s$ �j �� �jj� �fdd��}dS )z� Increase the the number of idle workers by C{n}. @param n: The number of new idle workers to create. @type n: L{int} c s2 t � �D ]} ��� }|d u r d S ��|� qd S r )�ranger �_recycleWorker)�x�worker��nr r r �createOneWorkeru s �z"Team.grow.<locals>.createOneWorkerN�r �checkr �do)r r* r+ r r) r �growl s z Team.growNc s$ �j �� �j�� �fdd�� dS )z� Decrease the number of idle workers by C{n}. @param n: The number of idle workers to shut down, or L{None} (or unspecified) to shut down all workers. @type n: L{int} or L{None} c s �� � �S r )� _quitIdlersr r) r r �<lambda>� � z Team.shrink.<locals>.<lambda>Nr, )r r* r r) r �shrink} s z Team.shrinkc C sp |du r t | j�| j }t|�D ]}| jr | j�� �� q| jd7 _q| jr4| jdkr6| j�� dS dS dS )z| The implmentation of C{shrink}, performed by the coordinator worker. @param n: see L{Team.shrink} Nr r ) r" r r r% �pop�quitr r r )r r* r' r r r r0 � s �zTeam._quitIdlersc s$ � j �� � j�� �fdd�� dS )zu Perform some work in a worker created by C{createWorker}. @param task: the callable to run c s � � ��S r )�_coordinateThisTaskr �r �taskr r r1 � r2 zTeam.do.<locals>.<lambda>Nr, r7 r r7 r r. � s zTeam.doc sX � j r� j �� n� �� ��du r� j��� dS � jd7 _�j� ��fdd��}dS )z� Select a worker to dispatch to, either an idle one or a new one, and perform it. This method should run on the coordinator worker. @param task: the task to dispatch @type task: 0-argument callable Nr c s@ z�� W n t y � �� Y nw � jj� �fdd��} d S )Nc s � j d8 _ � ��� d S )Nr )r r& r �r r( r r �idleAndPending� s z@Team._coordinateThisTask.<locals>.doWork.<locals>.idleAndPending)� BaseExceptionr r r. )r: �r r8 r( r r �doWork� s �z(Team._coordinateThisTask.<locals>.doWork)r r4 r r �appendr r. )r r8 r= r r< r r6 � s zTeam._coordinateThisTaskc C sl | j �|� | jr| �| j�� � dS | jr | �� dS | jdkr4| jd8 _| j �|� |� � dS dS )z� Called only from coordinator. Recycle the given worker into the idle pool. @param worker: a worker created by C{createWorker} and now idle. @type worker: L{IWorker} r r N) r �addr r6 �popleftr r0 r �remover5 r9 r r r r& � s �zTeam._recycleWorkerc s" � j �� � jj� fdd��}dS )zA Stop doing work and shut down all idle workers. c s d� _ � �� d S )NT)r r0 r r# r r �startFinishing� s z!Team.quit.<locals>.startFinishingN)r r r r. )r rB r r# r r5 � s z Team.quitr ) r r r r r r$ r/ r3 r0 r. r6 r& r5 r r r r r ) s r N) r � collectionsr �zope.interfacer � r � _conveniencer r r r r r r �<module> s __pycache__/_convenience.cpython-310.pyc 0000644 00000002302 15030470614 0014105 0 ustar 00 o �b~ � @ s" d Z ddlmZ G dd� d�ZdS )zI Common functionality used within the implementation of various workers. � )� AlreadyQuitc @ s( e Zd ZdZdd� Zdd� Zdd� ZdS ) �Quitz� A flag representing whether a worker has been quit. @ivar isSet: Whether this flag is set. @type isSet: L{bool} c C s d| _ dS )z* Create a L{Quit} un-set. FN)�isSet��self� r �?/usr/lib/python3/dist-packages/twisted/_threads/_convenience.py�__init__ s z Quit.__init__c C s | � � d| _dS )zg Set the flag if it has not been set. @raise AlreadyQuit: If it has been set. TN)�checkr r r r r �set s zQuit.setc C s | j rt� �dS )zb Check if the flag has been set. @raise AlreadyQuit: If it has been set. N)r r r r r r r $ s �z Quit.checkN)�__name__� __module__� __qualname__�__doc__r r r r r r r r s r N)r � _ithreadsr r r r r r �<module> s __pycache__/__init__.cpython-310.pyc 0000644 00000001021 15030470614 0013206 0 ustar 00 o �b� � @ sT d Z ddlmZmZ ddlmZ ddlmZ ddlm Z ddl m Z m Z g d�Z dS ) z4 Twisted integration with operating system threads. � )� AlreadyQuit�IWorker)�createMemoryWorker)�pool)�Team)� LockWorker� ThreadWorker)r r r r r r r N)�__doc__� _ithreadsr r �_memoryr �_poolr �_teamr � _threadworkerr r �__all__� r r �;/usr/lib/python3/dist-packages/twisted/_threads/__init__.py�<module> s __pycache__/_memory.cpython-310.pyc 0000644 00000003745 15030470614 0013135 0 ustar 00 o �b9 � @ sP d Z ddlmZ ddlmZ ddlmZ e� Zee�G dd� d��Z dd � Z d S ) z> Implementation of an in-memory worker that defers execution. � )� implementer� )�IWorker)�Quitc @ s, e Zd ZdZefdd�Zdd� Zdd� ZdS ) � MemoryWorkerz| An L{IWorker} that queues work for later performance. @ivar _quit: a flag indicating @type _quit: L{Quit} c C s t � | _|� | _dS )z+ Create a L{MemoryWorker}. N)r �_quit�_pending)�self�pending� r �:/usr/lib/python3/dist-packages/twisted/_threads/_memory.py�__init__ s zMemoryWorker.__init__c C s | j �� | j�|� dS )z} Queue some work for to perform later; see L{createMemoryWorker}. @param work: The work to perform. N)r �checkr �append)r �workr r r �do" s zMemoryWorker.doc C s | j �� | j�t� dS )z# Quit this worker. N)r �setr r � NoMoreWork)r r r r �quit+ s zMemoryWorker.quitN)�__name__� __module__� __qualname__�__doc__�listr r r r r r r r s r c s � fdd�} t � � � | fS )a Create an L{IWorker} that does nothing but defer work, to be performed later. @return: a worker that will enqueue work to perform later, and a callable that will perform one element of that work. @rtype: 2-L{tuple} of (L{IWorker}, L{callable}) c s. � j sdS � j d tu rdS � j �d�� dS )NFr T)r r �popr ��workerr r �perform= s z#createMemoryWorker.<locals>.perform)r )r r r r �createMemoryWorker3 s r N) r �zope.interfacer � r � _conveniencer �objectr r r r r r r �<module> s __pycache__/_pool.cpython-310.pyc 0000644 00000004745 15030470614 0012577 0 ustar 00 o �b� � @ s\ d Z ddlmZ ddlmZmZmZ ddlm Z ddl m Z ddl m Z mZ efdd �Zd S ) zR Top level thread pool interface, used to implement L{twisted.python.threadpool}. � )�Queue)�Lock�Thread�local)�err� )�Team)� LockWorker� ThreadWorkerc s8 �fdd��� ��fdd�}t tt� t� �|td���S )a; Construct a L{Team} that spawns threads as a thread pool, with the given limiting function. @note: Future maintainers: while the public API for the eventual move to twisted.threads should look I{something} like this, and while this function is necessary to implement the API described by L{twisted.python.threadpool}, I am starting to think the idea of a hard upper limit on threadpool size is just bad (turning memory performance issues into correctness issues well before we run into memory pressure), and instead we should build something with reactor integration for slowly releasing idle threads when they're not needed and I{rate} limiting the creation of new threads rather than just hard-capping it. @param currentLimit: a callable that returns the current limit on the number of workers that the returned L{Team} should create; if it already has more workers than that value, no new workers will be created. @type currentLimit: 0-argument callable returning L{int} @param threadFactory: Factory that, when given a C{target} keyword argument, returns a L{threading.Thread} that will run that target. @type threadFactory: callable returning a L{threading.Thread} @return: a new L{Team}. c s � | d�� � S )N��target)�startr )� threadFactory� �8/usr/lib/python3/dist-packages/twisted/_threads/_pool.py� startThread0 s zpool.<locals>.startThreadc s* �� � } | j| j � � krd S t�t� �S )N)� statistics�busyWorkerCount�idleWorkerCountr r )�stats)� currentLimitr �teamr r �limitedWorkerCreator3 s z"pool.<locals>.limitedWorkerCreator)� coordinator� createWorker� logException)r r r � LocalStorager )r r r r )r r r r r �pool s �r N)�__doc__�queuer � threadingr r r r �twisted.python.logr �_teamr � _threadworkerr r r r r r r �<module> s __pycache__/_ithreads.cpython-310.pyc 0000644 00000004332 15030470614 0013421 0 ustar 00 o �b� � @ sD d Z ddlmZ G dd� de�ZG dd� de�ZG dd� de�Zd S ) z Interfaces related to threads. � )� Interfacec @ � e Zd ZdZdS )� AlreadyQuitzJ This worker worker is dead and cannot execute more instructions. N��__name__� __module__� __qualname__�__doc__� r r �</usr/lib/python3/dist-packages/twisted/_threads/_ithreads.pyr � r c @ s e Zd ZdZdd� Zdd� ZdS )�IWorkerzs A worker that can perform some work concurrently. All methods on this interface must be thread-safe. c C � dS )a; Perform the given task. As an interface, this method makes no specific claims about concurrent execution. An L{IWorker}'s C{do} implementation may defer execution for later on the same thread, immediately on a different thread, or some combination of the two. It is valid for a C{do} method to schedule C{task} in such a way that it may never be executed. It is important for some implementations to provide specific properties with respect to where C{task} is executed, of course, and client code may rely on a more specific implementation of C{do} than L{IWorker}. @param task: a task to call in a thread or other concurrent context. @type task: 0-argument callable @raise AlreadyQuit: if C{quit} has been called. Nr )�taskr r r �do � z IWorker.doc C r )z� Free any resources associated with this L{IWorker} and cause it to reject all future work. @raise AlreadyQuit: if this method has already been called. Nr r r r r �quit. r z IWorker.quitN)r r r r r r r r r r r s r c @ r )�IExclusiveWorkerz� Like L{IWorker}, but with the additional guarantee that the callables passed to C{do} will not be called exclusively with each other. Nr r r r r r 7 r r N)r �zope.interfacer � Exceptionr r r r r r r �<module> s $ test/test_team.py 0000644 00000022367 15030470614 0010074 0 ustar 00 # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Tests for L{twisted._threads._team}. """ from twisted.python.components import proxyForInterface from twisted.python.context import call, get from twisted.python.failure import Failure from twisted.trial.unittest import SynchronousTestCase from .. import AlreadyQuit, IWorker, Team, createMemoryWorker class ContextualWorker(proxyForInterface(IWorker, "_realWorker")): # type: ignore[misc] """ A worker implementation that supplies a context. """ def __init__(self, realWorker, **ctx): """ Create with a real worker and a context. """ self._realWorker = realWorker self._context = ctx def do(self, work): """ Perform the given work with the context given to __init__. @param work: the work to pass on to the real worker. """ super().do(lambda: call(self._context, work)) class TeamTests(SynchronousTestCase): """ Tests for L{Team} """ def setUp(self): """ Set up a L{Team} with inspectable, synchronous workers that can be single-stepped. """ coordinator, self.coordinateOnce = createMemoryWorker() self.coordinator = ContextualWorker(coordinator, worker="coordinator") self.workerPerformers = [] self.allWorkersEver = [] self.allUnquitWorkers = [] self.activePerformers = [] self.noMoreWorkers = lambda: False def createWorker(): if self.noMoreWorkers(): return None worker, performer = createMemoryWorker() self.workerPerformers.append(performer) self.activePerformers.append(performer) cw = ContextualWorker(worker, worker=len(self.workerPerformers)) self.allWorkersEver.append(cw) self.allUnquitWorkers.append(cw) realQuit = cw.quit def quitAndRemove(): realQuit() self.allUnquitWorkers.remove(cw) self.activePerformers.remove(performer) cw.quit = quitAndRemove return cw self.failures = [] def logException(): self.failures.append(Failure()) self.team = Team(coordinator, createWorker, logException) def coordinate(self): """ Perform all work currently scheduled in the coordinator. @return: whether any coordination work was performed; if the coordinator was idle when this was called, return L{False} (otherwise L{True}). @rtype: L{bool} """ did = False while self.coordinateOnce(): did = True return did def performAllOutstandingWork(self): """ Perform all work on the coordinator and worker performers that needs to be done. """ continuing = True while continuing: continuing = self.coordinate() for performer in self.workerPerformers: if performer in self.activePerformers: performer() continuing = continuing or self.coordinate() def test_doDoesWorkInWorker(self): """ L{Team.do} does the work in a worker created by the createWorker callable. """ def something(): something.who = get("worker") self.team.do(something) self.coordinate() self.assertEqual(self.team.statistics().busyWorkerCount, 1) self.performAllOutstandingWork() self.assertEqual(something.who, 1) self.assertEqual(self.team.statistics().busyWorkerCount, 0) def test_initialStatistics(self): """ L{Team.statistics} returns an object with idleWorkerCount, busyWorkerCount, and backloggedWorkCount integer attributes. """ stats = self.team.statistics() self.assertEqual(stats.idleWorkerCount, 0) self.assertEqual(stats.busyWorkerCount, 0) self.assertEqual(stats.backloggedWorkCount, 0) def test_growCreatesIdleWorkers(self): """ L{Team.grow} increases the number of available idle workers. """ self.team.grow(5) self.performAllOutstandingWork() self.assertEqual(len(self.workerPerformers), 5) def test_growCreateLimit(self): """ L{Team.grow} increases the number of available idle workers until the C{createWorker} callable starts returning None. """ self.noMoreWorkers = lambda: len(self.allWorkersEver) >= 3 self.team.grow(5) self.performAllOutstandingWork() self.assertEqual(len(self.allWorkersEver), 3) self.assertEqual(self.team.statistics().idleWorkerCount, 3) def test_shrinkQuitsWorkers(self): """ L{Team.shrink} will quit the given number of workers. """ self.team.grow(5) self.performAllOutstandingWork() self.team.shrink(3) self.performAllOutstandingWork() self.assertEqual(len(self.allUnquitWorkers), 2) def test_shrinkToZero(self): """ L{Team.shrink} with no arguments will stop all outstanding workers. """ self.team.grow(10) self.performAllOutstandingWork() self.assertEqual(len(self.allUnquitWorkers), 10) self.team.shrink() self.assertEqual(len(self.allUnquitWorkers), 10) self.performAllOutstandingWork() self.assertEqual(len(self.allUnquitWorkers), 0) def test_moreWorkWhenNoWorkersAvailable(self): """ When no additional workers are available, the given work is backlogged, and then performed later when the work was. """ self.team.grow(3) self.coordinate() def something(): something.times += 1 something.times = 0 self.assertEqual(self.team.statistics().idleWorkerCount, 3) for i in range(3): self.team.do(something) # Make progress on the coordinator but do _not_ actually complete the # work, yet. self.coordinate() self.assertEqual(self.team.statistics().idleWorkerCount, 0) self.noMoreWorkers = lambda: True self.team.do(something) self.coordinate() self.assertEqual(self.team.statistics().idleWorkerCount, 0) self.assertEqual(self.team.statistics().backloggedWorkCount, 1) self.performAllOutstandingWork() self.assertEqual(self.team.statistics().backloggedWorkCount, 0) self.assertEqual(something.times, 4) def test_exceptionInTask(self): """ When an exception is raised in a task passed to L{Team.do}, the C{logException} given to the L{Team} at construction is invoked in the exception context. """ self.team.do(lambda: 1 / 0) self.performAllOutstandingWork() self.assertEqual(len(self.failures), 1) self.assertEqual(self.failures[0].type, ZeroDivisionError) def test_quit(self): """ L{Team.quit} causes future invocations of L{Team.do} and L{Team.quit} to raise L{AlreadyQuit}. """ self.team.quit() self.assertRaises(AlreadyQuit, self.team.quit) self.assertRaises(AlreadyQuit, self.team.do, list) def test_quitQuits(self): """ L{Team.quit} causes all idle workers, as well as the coordinator worker, to quit. """ for x in range(10): self.team.do(list) self.performAllOutstandingWork() self.team.quit() self.performAllOutstandingWork() self.assertEqual(len(self.allUnquitWorkers), 0) self.assertRaises(AlreadyQuit, self.coordinator.quit) def test_quitQuitsLaterWhenBusy(self): """ L{Team.quit} causes all busy workers to be quit once they've finished the work they've been given. """ self.team.grow(10) for x in range(5): self.team.do(list) self.coordinate() self.team.quit() self.coordinate() self.assertEqual(len(self.allUnquitWorkers), 5) self.performAllOutstandingWork() self.assertEqual(len(self.allUnquitWorkers), 0) self.assertRaises(AlreadyQuit, self.coordinator.quit) def test_quitConcurrentWithWorkHappening(self): """ If work happens after L{Team.quit} sets its C{Quit} flag, but before any other work takes place, the L{Team} should still exit gracefully. """ self.team.do(list) originalSet = self.team._quit.set def performWorkConcurrently(): originalSet() self.performAllOutstandingWork() self.team._quit.set = performWorkConcurrently self.team.quit() self.assertRaises(AlreadyQuit, self.team.quit) self.assertRaises(AlreadyQuit, self.team.do, list) def test_shrinkWhenBusy(self): """ L{Team.shrink} will wait for busy workers to finish being busy and then quit them. """ for x in range(10): self.team.do(list) self.coordinate() self.assertEqual(len(self.allUnquitWorkers), 10) # There should be 10 busy workers at this point. self.team.shrink(7) self.performAllOutstandingWork() self.assertEqual(len(self.allUnquitWorkers), 3) test/test_convenience.py 0000644 00000002534 15030470614 0011434 0 ustar 00 # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Test cases for convenience functionality in L{twisted._threads._convenience}. """ from twisted.trial.unittest import SynchronousTestCase from .._convenience import Quit from .._ithreads import AlreadyQuit class QuitTests(SynchronousTestCase): """ Tests for L{Quit} """ def test_isInitiallySet(self): """ L{Quit.isSet} starts as L{False}. """ quit = Quit() self.assertEqual(quit.isSet, False) def test_setSetsSet(self): """ L{Quit.set} sets L{Quit.isSet} to L{True}. """ quit = Quit() quit.set() self.assertEqual(quit.isSet, True) def test_checkDoesNothing(self): """ L{Quit.check} initially does nothing and returns L{None}. """ quit = Quit() self.assertIs(quit.check(), None) def test_checkAfterSetRaises(self): """ L{Quit.check} raises L{AlreadyQuit} if L{Quit.set} has been called. """ quit = Quit() quit.set() self.assertRaises(AlreadyQuit, quit.check) def test_setTwiceRaises(self): """ L{Quit.set} raises L{AlreadyQuit} if it has been called previously. """ quit = Quit() quit.set() self.assertRaises(AlreadyQuit, quit.set) test/__pycache__/test_convenience.cpython-310.pyc 0000644 00000003617 15030470614 0015776 0 ustar 00 o �b\ � @ s< d Z ddlmZ ddlmZ ddlmZ G dd� de�ZdS ) zO Test cases for convenience functionality in L{twisted._threads._convenience}. � )�SynchronousTestCase� )�Quit)� AlreadyQuitc @ s8 e Zd ZdZdd� Zdd� Zdd� Zdd � Zd d � Zd S ) � QuitTestsz Tests for L{Quit} c C s t � }| �|jd� dS )z3 L{Quit.isSet} starts as L{False}. FN)r � assertEqual�isSet��self�quit� r �H/usr/lib/python3/dist-packages/twisted/_threads/test/test_convenience.py�test_isInitiallySet s z QuitTests.test_isInitiallySetc C s t � }|�� | �|jd� dS )z< L{Quit.set} sets L{Quit.isSet} to L{True}. TN)r �setr r r r r r �test_setSetsSet � zQuitTests.test_setSetsSetc C s t � }| �|�� d� dS )zK L{Quit.check} initially does nothing and returns L{None}. N)r �assertIs�checkr r r r �test_checkDoesNothing" s z QuitTests.test_checkDoesNothingc C s t � }|�� | �t|j� dS )zU L{Quit.check} raises L{AlreadyQuit} if L{Quit.set} has been called. N)r r � assertRaisesr r r r r r �test_checkAfterSetRaises) r z"QuitTests.test_checkAfterSetRaisesc C s t � }|�� | �t|j� dS )zU L{Quit.set} raises L{AlreadyQuit} if it has been called previously. N)r r r r r r r r �test_setTwiceRaises1 r z QuitTests.test_setTwiceRaisesN) �__name__� __module__� __qualname__�__doc__r r r r r r r r r r s r N)r �twisted.trial.unittestr � _conveniencer � _ithreadsr r r r r r �<module> s test/__pycache__/__init__.cpython-310.pyc 0000644 00000000311 15030470614 0014166 0 ustar 00 o �b� � @ s d Z dS )z Tests for L{twisted._threads}. N)�__doc__� r r �@/usr/lib/python3/dist-packages/twisted/_threads/test/__init__.py�<module> s test/__pycache__/test_memory.cpython-310.pyc 0000644 00000005102 15030470614 0015001 0 ustar 00 o �b# � @ sD d Z ddlmZ ddlmZ ddlmZmZmZ G dd� de�Z dS ) z( Tests for L{twisted._threads._memory}. � )� verifyObject)�SynchronousTestCase� )� AlreadyQuit�IWorker�createMemoryWorkerc @ s( e Zd ZdZdd� Zdd� Zdd� ZdS ) �MemoryWorkerTestsz$ Tests for L{MemoryWorker}. c s� t � \}}tt|� g � |�� fdd�� |�� fdd�� | �� g � | �|� d� | �� dg� | �|� d� | �� ddg� dS )z� L{createMemoryWorker} creates an L{IWorker} and a callable that can perform work on it. The performer returns C{True} if it accomplished useful work. c � � � d�S )N� ��append� ��doner �C/usr/lib/python3/dist-packages/twisted/_threads/test/test_memory.py�<lambda> � z?MemoryWorkerTests.test_createWorkerAndPerform.<locals>.<lambda>c r )N� r r r r r r r Tr r N)r r r �do� assertEqual��self�worker� performerr r r �test_createWorkerAndPerform s z-MemoryWorkerTests.test_createWorkerAndPerformc sp t � \}}g � � fdd�}|�|� |�� | �t|j|� | �t|j� |� | �� dg� | �|� d� dS )a Calling C{quit} on the worker returned by L{createMemoryWorker} causes its C{do} and C{quit} methods to raise L{AlreadyQuit}; its C{perform} callable will start raising L{AlreadyQuit} when the work already provided to C{do} has been exhausted. c s � � d� d S )N� r r r r r �moreWork. s z2MemoryWorkerTests.test_quitQuits.<locals>.moreWorkr FN)r r �quit� assertRaisesr r )r r r r r r r �test_quitQuits$ s z MemoryWorkerTests.test_quitQuitsc C s t � \}}| �|� d� dS )z� The C{perform} callable returned by L{createMemoryWorker} will return no result when there's no work to do yet. Since there is no work to do, the performer returns C{False}. FN)r r r r r r � test_performWhenNothingToDoYet9 s z0MemoryWorkerTests.test_performWhenNothingToDoYetN)�__name__� __module__� __qualname__�__doc__r r r r r r r r s r N) r$ �zope.interface.verifyr �twisted.trial.unittestr � r r r r r r r r �<module> s test/__pycache__/test_threadworker.cpython-310.pyc 0000644 00000023265 15030470614 0016204 0 ustar 00 o �b1 � @ s� d Z ddlZddlZddlmZmZ ddlmZ ddlm Z m Z m Z G dd� de �Z G d d � d e �ZG d d � d �ZG d d� d�ZG dd� d�ZG dd� de�ZG dd� de�ZdS )z. Tests for L{twisted._threads._threadworker}. � N)� ThreadError�local)�SynchronousTestCase� )� AlreadyQuit� LockWorker� ThreadWorkerc @ � e Zd ZdZdS )�FakeQueueEmptyz8 L{FakeQueue}'s C{get} has exhausted the queue. N��__name__� __module__� __qualname__�__doc__� r r �I/usr/lib/python3/dist-packages/twisted/_threads/test/test_threadworker.pyr � r c @ r )� WouldDeadlockzf If this were a real lock, you'd be deadlocked because the lock would be double-acquired. Nr r r r r r r r c @ s e Zd ZdZdd� Zdd� ZdS )� FakeThreadz� A fake L{threading.Thread}. @ivar target: A target function to run. @type target: L{callable} @ivar started: Has this thread been started? @type started: L{bool} c C s || _ d| _dS )z7 Create a L{FakeThread} with a target. FN)�target�started)�selfr r r r �__init__) s zFakeThread.__init__c C � d| _ dS )z) Set the "started" flag. TN)r �r r r r �start0 � zFakeThread.startN)r r r r r r r r r r r s r c @ �( e Zd ZdZdd� Zdd� Zdd� ZdS ) � FakeQueuez� A fake L{Queue} implementing C{put} and C{get}. @ivar items: A lit of items placed by C{put} but not yet retrieved by C{get}. @type items: L{list} c C s g | _ dS )z( Create a L{FakeQueue}. N)�itemsr r r r r @ r zFakeQueue.__init__c C s | j �|� dS )zv Put an item into the queue for later retrieval by L{FakeQueue.get}. @param item: any object N)r �append)r �itemr r r �putF s z FakeQueue.putc C s | j st� �| j �d�S )zR Get an item. @return: an item previously put by C{put}. r )r r �popr r r r �getN s z FakeQueue.getN)r r r r r r" r$ r r r r r 7 s r c @ r ) �FakeLockzi A stand-in for L{threading.Lock}. @ivar acquired: Whether this lock is presently acquired. c C r )z9 Create a lock in the un-acquired state. FN)�acquiredr r r r r ` r zFakeLock.__init__c C s | j rt� �d| _ dS )zX Acquire the lock. Raise an exception if the lock is already acquired. TN)r& r r r r r �acquiref s zFakeLock.acquirec C s | j st� �d| _ dS )zf Release the lock. Raise an exception if the lock is not presently acquired. FN)r& r r r r r �releasen s zFakeLock.releaseN)r r r r r r' r( r r r r r% Y s r% c @ r ) �ThreadWorkerTestsz$ Tests for L{ThreadWorker}. c s, g � _ t� � _� fdd�}t|� j�� _dS )z4 Create a worker with fake threads. c s" t | d�}|�� � j�|� |S )N)r )r r � fakeThreadsr )r � newThreadr r r � startThread� s z,ThreadWorkerTests.setUp.<locals>.startThreadN)r* r � fakeQueuer �worker)r r, r r r �setUp} s zThreadWorkerTests.setUpc sx | � t| j�d� | � | jd jd� � fdd�� d� _| j�� � | � � jd� | �t| jd j � | � � jd� dS )z� L{ThreadWorker} calls its C{createThread} callable to create a thread, its C{createQueue} callable to create a queue, and then the thread's target pulls work from that queue. � r Tc s d� _ d S �NT)�doner ��doItr r r4 � s z@ThreadWorkerTests.test_startsThreadAndPerformsWork.<locals>.doItFN) � assertEqual�lenr* r r2 r. �do� assertRaisesr r r r r3 r � test_startsThreadAndPerformsWork� s z2ThreadWorkerTests.test_startsThreadAndPerformsWorkc C s0 | j �� | �t| j j� | �t| j jt� dS )z� L{ThreadWorker.quit} causes future calls to L{ThreadWorker.do} and L{ThreadWorker.quit} to raise L{AlreadyQuit}. N)r. �quitr8 r r7 �listr r r r � test_quitPreventsFutureCalls� s z.ThreadWorkerTests.test_quitPreventsFutureCallsN)r r r r r/ r9 r< r r r r r) x s r) c @ sH e Zd ZdZdd� Zdd� Zdd� Zdd � Zd d � Zd d � Z dd� Z dS )�LockWorkerTestsz" Tests for L{LockWorker}. c C s t � }|�� | �t|j� dS )ze The L{FakeLock} test fixture will alert us if there's a potential deadlock. N)r% r' r8 r �r �lockr r r �test_fakeDeadlock� s z!LockWorkerTests.test_fakeDeadlockc C s> t � }| �t|j� |�� | �d|�� � | �t|j� dS )zk The L{FakeLock} test fixture will alert us if there's a potential double-release. N)r% r8 r r( r' r5 r> r r r �test_fakeDoubleRelease� s z&LockWorkerTests.test_fakeDoubleReleasec sb t � }t� � t� |�}� �fdd��d�_|��� | ��jd� | ��jd� | �� jd� dS )zp L{LockWorker.do} immediately performs the work it's given, while the lock is acquired. c � d�_ � j�_d S r1 )r2 r& r �r? �workr r rD � � z@LockWorkerTests.test_doExecutesImmediatelyWithLock.<locals>.workFTN)r r% r r2 r7 r5 r& )r �storager. r rC r �"test_doExecutesImmediatelyWithLock� s z2LockWorkerTests.test_doExecutesImmediatelyWithLockc sb t � �t�t� ��g �g � � ����fdd��d�_���� | ��ddg� | �� ddg� dS )z� If L{LockWorker.do} is called recursively, it postpones the inner call until the outer one is complete. c sN � j d7 _ ���j � � ��j� t��dk r ���� � j d8 _ d S )Nr0 r )�levelr r& r6 r7 r �r& �levelsr? rD r. r r rD � s z6LockWorkerTests.test_doUnwindsReentrancy.<locals>.workr r0 TN)r% r r rH r7 r5 r r rI r �test_doUnwindsReentrancy� s z(LockWorkerTests.test_doUnwindsReentrancyc C sn t � }t�|�}t|t� �}d}| �|� d� |�� t�� | � |� d� | � t |j� | � t |j t � dS )z� L{LockWorker.quit} frees the resources associated with its lock and causes further calls to C{do} and C{quit} to fail. N)r% �weakref�refr r � assertIsNotr: �gc�collect�assertIsr8 r r7 r; )r r? rM r. r r r � test_quit� s zLockWorkerTests.test_quitc s� t � � t�� �}t� t� ������fdd��d�_� �fdd��d�_���� ���jd� ���jd� ��� jd� d� t � � �� |� d� dS )z� If L{LockWorker.quit} is invoked during a call to L{LockWorker.do}, all recursive work scheduled with L{LockWorker.do} will be completed and the lock will be released. c s, �� �� ��� ��t�j t� d� _d S r1 )r7 r: r8 r r; �completer )�phase1�phase2r r. r r rT s z5LockWorkerTests.test_quitWhileWorking.<locals>.phase1Fc rB r1 )rS r& r )r? rU r r rU rE z5LockWorkerTests.test_quitWhileWorking.<locals>.phase2TN) r% rL rM r r rS r7 r5 r& rO rP rQ )r rM r )r? rT rU r r. r �test_quitWhileWorking� s z%LockWorkerTests.test_quitWhileWorkingc C s2 G dd� dt �}|t� t� �}| �t|jt� dS )z� If L{LockWorker.do} is called concurrently with L{LockWorker.quit}, and C{quit} wins the race before C{do} gets the lock attribute, then L{AlreadyQuit} will be raised. c @ s& e Zd Zedd� �Zejdd� �ZdS )zALockWorkerTests.test_quitWhileGettingLock.<locals>.RacyLockWorkerc S s | � � | jd S �N�_lock)r: �__dict__r r r r rX " s zGLockWorkerTests.test_quitWhileGettingLock.<locals>.RacyLockWorker._lockc S s || j d< d S rW )rY )r �valuer r r rX ' s N)r r r �propertyrX �setterr r r r �RacyLockWorker! s r] N)r r% r r8 r r7 r; )r r] r. r r r �test_quitWhileGettingLock s z)LockWorkerTests.test_quitWhileGettingLockN) r r r r r@ rA rG rK rR rV r^ r r r r r= � s r= )r rO rL � threadingr r �twisted.trial.unittestr � r r r � Exceptionr r r r r% r) r= r r r r �<module> s " 0 test/__pycache__/test_team.cpython-310.pyc 0000644 00000024213 15030470614 0014423 0 ustar 00 o �b�$ � @ sz d Z ddlmZ ddlmZmZ ddlmZ ddlm Z ddl m Z m Z m Z mZ G dd � d ee d ��ZG d d � d e �Zd S )z& Tests for L{twisted._threads._team}. � )�proxyForInterface)�call�get)�Failure)�SynchronousTestCase� )� AlreadyQuit�IWorker�Team�createMemoryWorkerc s( e Zd ZdZdd� Z� fdd�Z� ZS )�ContextualWorkerz: A worker implementation that supplies a context. c K s || _ || _dS )z: Create with a real worker and a context. N)� _realWorker�_context)�self� realWorker�ctx� r �A/usr/lib/python3/dist-packages/twisted/_threads/test/test_team.py�__init__ s zContextualWorker.__init__c s t � �� �fdd�� dS )z� Perform the given work with the context given to __init__. @param work: the work to pass on to the real worker. c s t � j��S �N)r r r �r �workr r �<lambda>" s z%ContextualWorker.do.<locals>.<lambda>N)�super�dor �� __class__r r r s zContextualWorker.do)�__name__� __module__� __qualname__�__doc__r r � __classcell__r r r r r s r 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"S )#� TeamTestsz Tests for L{Team} c sl t � \}� _t|dd�� _g � _g � _g � _g � _dd� � _� fdd�}g � _ � fdd�}t |||�� _ d S ) zl Set up a L{Team} with inspectable, synchronous workers that can be single-stepped. � coordinator��workerc S � dS )NFr r r r r r 5 � z!TeamTests.setUp.<locals>.<lambda>c sz �� � rd S t� \} ��j��� �j��� t| t�j�d�� �j�� � �j�� � � j �� ���fdd�}|� _ � S )Nr$ c s" �� �j �� � �j��� d S r )�allUnquitWorkers�remove�activePerformersr )�cw� performer�realQuitr r r � quitAndRemoveB s z<TeamTests.setUp.<locals>.createWorker.<locals>.quitAndRemove) � noMoreWorkersr �workerPerformers�appendr* r �len�allWorkersEverr( �quit)r% r. �r )r+ r, r- r � createWorker7 s z%TeamTests.setUp.<locals>.createWorkerc s � j �t� � d S r )�failuresr1 r r r5 r r � logExceptionL � z%TeamTests.setUp.<locals>.logExceptionN) r �coordinateOncer r# r0 r3 r( r* r/ r7 r �team)r r# r6 r8 r r5 r �setUp* s zTeamTests.setUpc C s d}| � � r d}| � � s|S )a Perform all work currently scheduled in the coordinator. @return: whether any coordination work was performed; if the coordinator was idle when this was called, return L{False} (otherwise L{True}). @rtype: L{bool} FT)r: )r �didr r r � coordinateQ s �zTeamTests.coordinatec C sD d}|r | � � }| jD ] }|| jv r|� q |p| � � }|sdS dS )zj Perform all work on the coordinator and worker performers that needs to be done. TN)r> r0 r* )r � continuingr, r r r �performAllOutstandingWork_ s � �z#TeamTests.performAllOutstandingWorkc sb � fdd�� | j �� � | �� | �| j �� jd� | �� | �� jd� | �| j �� jd� dS )zd L{Team.do} does the work in a worker created by the createWorker callable. c s t d�� _d S )Nr% )r �whor �� somethingr r rC r s z4TeamTests.test_doDoesWorkInWorker.<locals>.something� r N)r; r r> � assertEqual� statistics�busyWorkerCountr@ rA r5 r rB r �test_doDoesWorkInWorkerl s z!TeamTests.test_doDoesWorkInWorkerc C s8 | j �� }| �|jd� | �|jd� | �|jd� dS )z� L{Team.statistics} returns an object with idleWorkerCount, busyWorkerCount, and backloggedWorkCount integer attributes. r N)r; rF rE �idleWorkerCountrG �backloggedWorkCount)r �statsr r r �test_initialStatistics| s z TeamTests.test_initialStatisticsc C s* | j �d� | �� | �t| j�d� dS )zN L{Team.grow} increases the number of available idle workers. � N)r; �growr@ rE r2 r0 r5 r r r �test_growCreatesIdleWorkers� s z%TeamTests.test_growCreatesIdleWorkersc sL � fdd�� _ � j�d� � �� � �t� j�d� � �� j�� jd� dS )z� L{Team.grow} increases the number of available idle workers until the C{createWorker} callable starts returning None. c s t � j�dkS )N� )r2 r3 r r5 r r r � s z0TeamTests.test_growCreateLimit.<locals>.<lambda>rM rP N) r/ r; rN r@ rE r2 r3 rF rI r5 r r5 r �test_growCreateLimit� s z TeamTests.test_growCreateLimitc C s> | j �d� | �� | j �d� | �� | �t| j�d� dS )zG L{Team.shrink} will quit the given number of workers. rM rP r N)r; rN r@ �shrinkrE r2 r( r5 r r r �test_shrinkQuitsWorkers� s z!TeamTests.test_shrinkQuitsWorkersc C s` | j �d� | �� | �t| j�d� | j �� | �t| j�d� | �� | �t| j�d� dS )zU L{Team.shrink} with no arguments will stop all outstanding workers. � r N)r; rN r@ rE r2 r( rR r5 r r r �test_shrinkToZero� s zTeamTests.test_shrinkToZeroc s� | j �d� | �� � fdd�� d� _| �| j �� jd� td�D ]}| j �� � q!| �� | �| j �� jd� dd� | _ | j �� � | �� | �| j �� jd� | �| j �� j d� | � � | �| j �� j d� | �� jd� d S ) z� When no additional workers are available, the given work is backlogged, and then performed later when the work was. rP c s � j d7 _ d S )NrD )�timesr rB r r rC � r9 z@TeamTests.test_moreWorkWhenNoWorkersAvailable.<locals>.somethingr c S r& )NTr r r r r r � r' z?TeamTests.test_moreWorkWhenNoWorkersAvailable.<locals>.<lambda>rD � N) r; rN r> rV rE rF rI �ranger r/ rJ r@ )r �ir rB r �#test_moreWorkWhenNoWorkersAvailable� s" z-TeamTests.test_moreWorkWhenNoWorkersAvailablec C sB | j �dd� � | �� | �t| j�d� | �| jd jt� dS )z� When an exception is raised in a task passed to L{Team.do}, the C{logException} given to the L{Team} at construction is invoked in the exception context. c S s dd S )NrD r r r r r r r � s z0TeamTests.test_exceptionInTask.<locals>.<lambda>rD r N)r; r r@ rE r2 r7 �type�ZeroDivisionErrorr5 r r r �test_exceptionInTask� s z TeamTests.test_exceptionInTaskc C s0 | j �� | �t| j j� | �t| j jt� dS )zx L{Team.quit} causes future invocations of L{Team.do} and L{Team.quit} to raise L{AlreadyQuit}. N)r; r4 � assertRaisesr r �listr5 r r r � test_quit� s zTeamTests.test_quitc C sZ t d�D ]}| j�t� q| �� | j�� | �� | �t| j�d� | � t | j j� dS )zk L{Team.quit} causes all idle workers, as well as the coordinator worker, to quit. rT r N) rX r; r r_ r@ r4 rE r2 r( r^ r r# �r �xr r r �test_quitQuits� s zTeamTests.test_quitQuitsc C s� | j �d� td�D ]}| j �t� q | �� | j �� | �� | �t| j �d� | � � | �t| j �d� | � t | j j� dS )z| L{Team.quit} causes all busy workers to be quit once they've finished the work they've been given. rT rM r N)r; rN rX r r_ r> r4 rE r2 r( r@ r^ r r# ra r r r �test_quitQuitsLaterWhenBusy� s z%TeamTests.test_quitQuitsLaterWhenBusyc s^ �j �t� �j jj� � �fdd�}|�j j_�j �� ��t�j j� ��t�j jt� dS )z� If work happens after L{Team.quit} sets its C{Quit} flag, but before any other work takes place, the L{Team} should still exit gracefully. c s � � �� � d S r )r@ r �� originalSetr r r �performWorkConcurrently s zOTeamTests.test_quitConcurrentWithWorkHappening.<locals>.performWorkConcurrentlyN)r; r r_ �_quit�setr4 r^ r )r rg r re r �$test_quitConcurrentWithWorkHappening� s z.TeamTests.test_quitConcurrentWithWorkHappeningc C s^ t d�D ]}| j�t� q| �� | �t| j�d� | j�d� | � � | �t| j�d� dS )zl L{Team.shrink} will wait for busy workers to finish being busy and then quit them. rT � rP N) rX r; r r_ r> rE r2 r( rR r@ ra r r r �test_shrinkWhenBusy s z TeamTests.test_shrinkWhenBusyN)r r r r r< r> r@ rH rL rO rQ rS rU rZ r] r` rc rd rj rl r r r r r" % s$ ' r" N)r �twisted.python.componentsr �twisted.python.contextr r �twisted.python.failurer �twisted.trial.unittestr � r r r r r r" r r r r �<module> s test/test_threadworker.py 0000644 00000017461 15030470614 0011646 0 ustar 00 # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Tests for L{twisted._threads._threadworker}. """ import gc import weakref from threading import ThreadError, local from twisted.trial.unittest import SynchronousTestCase from .. import AlreadyQuit, LockWorker, ThreadWorker class FakeQueueEmpty(Exception): """ L{FakeQueue}'s C{get} has exhausted the queue. """ class WouldDeadlock(Exception): """ If this were a real lock, you'd be deadlocked because the lock would be double-acquired. """ class FakeThread: """ A fake L{threading.Thread}. @ivar target: A target function to run. @type target: L{callable} @ivar started: Has this thread been started? @type started: L{bool} """ def __init__(self, target): """ Create a L{FakeThread} with a target. """ self.target = target self.started = False def start(self): """ Set the "started" flag. """ self.started = True class FakeQueue: """ A fake L{Queue} implementing C{put} and C{get}. @ivar items: A lit of items placed by C{put} but not yet retrieved by C{get}. @type items: L{list} """ def __init__(self): """ Create a L{FakeQueue}. """ self.items = [] def put(self, item): """ Put an item into the queue for later retrieval by L{FakeQueue.get}. @param item: any object """ self.items.append(item) def get(self): """ Get an item. @return: an item previously put by C{put}. """ if not self.items: raise FakeQueueEmpty() return self.items.pop(0) class FakeLock: """ A stand-in for L{threading.Lock}. @ivar acquired: Whether this lock is presently acquired. """ def __init__(self): """ Create a lock in the un-acquired state. """ self.acquired = False def acquire(self): """ Acquire the lock. Raise an exception if the lock is already acquired. """ if self.acquired: raise WouldDeadlock() self.acquired = True def release(self): """ Release the lock. Raise an exception if the lock is not presently acquired. """ if not self.acquired: raise ThreadError() self.acquired = False class ThreadWorkerTests(SynchronousTestCase): """ Tests for L{ThreadWorker}. """ def setUp(self): """ Create a worker with fake threads. """ self.fakeThreads = [] self.fakeQueue = FakeQueue() def startThread(target): newThread = FakeThread(target=target) newThread.start() self.fakeThreads.append(newThread) return newThread self.worker = ThreadWorker(startThread, self.fakeQueue) def test_startsThreadAndPerformsWork(self): """ L{ThreadWorker} calls its C{createThread} callable to create a thread, its C{createQueue} callable to create a queue, and then the thread's target pulls work from that queue. """ self.assertEqual(len(self.fakeThreads), 1) self.assertEqual(self.fakeThreads[0].started, True) def doIt(): doIt.done = True doIt.done = False self.worker.do(doIt) self.assertEqual(doIt.done, False) self.assertRaises(FakeQueueEmpty, self.fakeThreads[0].target) self.assertEqual(doIt.done, True) def test_quitPreventsFutureCalls(self): """ L{ThreadWorker.quit} causes future calls to L{ThreadWorker.do} and L{ThreadWorker.quit} to raise L{AlreadyQuit}. """ self.worker.quit() self.assertRaises(AlreadyQuit, self.worker.quit) self.assertRaises(AlreadyQuit, self.worker.do, list) class LockWorkerTests(SynchronousTestCase): """ Tests for L{LockWorker}. """ def test_fakeDeadlock(self): """ The L{FakeLock} test fixture will alert us if there's a potential deadlock. """ lock = FakeLock() lock.acquire() self.assertRaises(WouldDeadlock, lock.acquire) def test_fakeDoubleRelease(self): """ The L{FakeLock} test fixture will alert us if there's a potential double-release. """ lock = FakeLock() self.assertRaises(ThreadError, lock.release) lock.acquire() self.assertEqual(None, lock.release()) self.assertRaises(ThreadError, lock.release) def test_doExecutesImmediatelyWithLock(self): """ L{LockWorker.do} immediately performs the work it's given, while the lock is acquired. """ storage = local() lock = FakeLock() worker = LockWorker(lock, storage) def work(): work.done = True work.acquired = lock.acquired work.done = False worker.do(work) self.assertEqual(work.done, True) self.assertEqual(work.acquired, True) self.assertEqual(lock.acquired, False) def test_doUnwindsReentrancy(self): """ If L{LockWorker.do} is called recursively, it postpones the inner call until the outer one is complete. """ lock = FakeLock() worker = LockWorker(lock, local()) levels = [] acquired = [] def work(): work.level += 1 levels.append(work.level) acquired.append(lock.acquired) if len(levels) < 2: worker.do(work) work.level -= 1 work.level = 0 worker.do(work) self.assertEqual(levels, [1, 1]) self.assertEqual(acquired, [True, True]) def test_quit(self): """ L{LockWorker.quit} frees the resources associated with its lock and causes further calls to C{do} and C{quit} to fail. """ lock = FakeLock() ref = weakref.ref(lock) worker = LockWorker(lock, local()) lock = None self.assertIsNot(ref(), None) worker.quit() gc.collect() self.assertIs(ref(), None) self.assertRaises(AlreadyQuit, worker.quit) self.assertRaises(AlreadyQuit, worker.do, list) def test_quitWhileWorking(self): """ If L{LockWorker.quit} is invoked during a call to L{LockWorker.do}, all recursive work scheduled with L{LockWorker.do} will be completed and the lock will be released. """ lock = FakeLock() ref = weakref.ref(lock) worker = LockWorker(lock, local()) def phase1(): worker.do(phase2) worker.quit() self.assertRaises(AlreadyQuit, worker.do, list) phase1.complete = True phase1.complete = False def phase2(): phase2.complete = True phase2.acquired = lock.acquired phase2.complete = False worker.do(phase1) self.assertEqual(phase1.complete, True) self.assertEqual(phase2.complete, True) self.assertEqual(lock.acquired, False) lock = None gc.collect() self.assertIs(ref(), None) def test_quitWhileGettingLock(self): """ If L{LockWorker.do} is called concurrently with L{LockWorker.quit}, and C{quit} wins the race before C{do} gets the lock attribute, then L{AlreadyQuit} will be raised. """ class RacyLockWorker(LockWorker): @property def _lock(self): self.quit() return self.__dict__["_lock"] @_lock.setter def _lock(self, value): self.__dict__["_lock"] = value worker = RacyLockWorker(FakeLock(), local()) self.assertRaises(AlreadyQuit, worker.do, list) test/test_memory.py 0000644 00000004043 15030470614 0010445 0 ustar 00 # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Tests for L{twisted._threads._memory}. """ from zope.interface.verify import verifyObject from twisted.trial.unittest import SynchronousTestCase from .. import AlreadyQuit, IWorker, createMemoryWorker class MemoryWorkerTests(SynchronousTestCase): """ Tests for L{MemoryWorker}. """ def test_createWorkerAndPerform(self): """ L{createMemoryWorker} creates an L{IWorker} and a callable that can perform work on it. The performer returns C{True} if it accomplished useful work. """ worker, performer = createMemoryWorker() verifyObject(IWorker, worker) done = [] worker.do(lambda: done.append(3)) worker.do(lambda: done.append(4)) self.assertEqual(done, []) self.assertEqual(performer(), True) self.assertEqual(done, [3]) self.assertEqual(performer(), True) self.assertEqual(done, [3, 4]) def test_quitQuits(self): """ Calling C{quit} on the worker returned by L{createMemoryWorker} causes its C{do} and C{quit} methods to raise L{AlreadyQuit}; its C{perform} callable will start raising L{AlreadyQuit} when the work already provided to C{do} has been exhausted. """ worker, performer = createMemoryWorker() done = [] def moreWork(): done.append(7) worker.do(moreWork) worker.quit() self.assertRaises(AlreadyQuit, worker.do, moreWork) self.assertRaises(AlreadyQuit, worker.quit) performer() self.assertEqual(done, [7]) self.assertEqual(performer(), False) def test_performWhenNothingToDoYet(self): """ The C{perform} callable returned by L{createMemoryWorker} will return no result when there's no work to do yet. Since there is no work to do, the performer returns C{False}. """ worker, performer = createMemoryWorker() self.assertEqual(performer(), False) test/__init__.py 0000644 00000000240 15030470614 0007630 0 ustar 00 # -*- test-case-name: twisted._threads.test -*- # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Tests for L{twisted._threads}. """ _team.py 0000644 00000015755 15030470614 0006220 0 ustar 00 # -*- test-case-name: twisted._threads.test.test_team -*- # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Implementation of a L{Team} of workers; a thread-pool that can allocate work to workers. """ from collections import deque from zope.interface import implementer from . import IWorker from ._convenience import Quit class Statistics: """ Statistics about a L{Team}'s current activity. @ivar idleWorkerCount: The number of idle workers. @type idleWorkerCount: L{int} @ivar busyWorkerCount: The number of busy workers. @type busyWorkerCount: L{int} @ivar backloggedWorkCount: The number of work items passed to L{Team.do} which have not yet been sent to a worker to be performed because not enough workers are available. @type backloggedWorkCount: L{int} """ def __init__(self, idleWorkerCount, busyWorkerCount, backloggedWorkCount): self.idleWorkerCount = idleWorkerCount self.busyWorkerCount = busyWorkerCount self.backloggedWorkCount = backloggedWorkCount @implementer(IWorker) class Team: """ A composite L{IWorker} implementation. @ivar _quit: A L{Quit} flag indicating whether this L{Team} has been quit yet. This may be set by an arbitrary thread since L{Team.quit} may be called from anywhere. @ivar _coordinator: the L{IExclusiveWorker} coordinating access to this L{Team}'s internal resources. @ivar _createWorker: a callable that will create new workers. @ivar _logException: a 0-argument callable called in an exception context when there is an unhandled error from a task passed to L{Team.do} @ivar _idle: a L{set} of idle workers. @ivar _busyCount: the number of workers currently busy. @ivar _pending: a C{deque} of tasks - that is, 0-argument callables passed to L{Team.do} - that are outstanding. @ivar _shouldQuitCoordinator: A flag indicating that the coordinator should be quit at the next available opportunity. Unlike L{Team._quit}, this flag is only set by the coordinator. @ivar _toShrink: the number of workers to shrink this L{Team} by at the next available opportunity; set in the coordinator. """ def __init__(self, coordinator, createWorker, logException): """ @param coordinator: an L{IExclusiveWorker} which will coordinate access to resources on this L{Team}; that is to say, an L{IExclusiveWorker} whose C{do} method ensures that its given work will be executed in a mutually exclusive context, not in parallel with other work enqueued by C{do} (although possibly in parallel with the caller). @param createWorker: A 0-argument callable that will create an L{IWorker} to perform work. @param logException: A 0-argument callable called in an exception context when the work passed to C{do} raises an exception. """ self._quit = Quit() self._coordinator = coordinator self._createWorker = createWorker self._logException = logException # Don't touch these except from the coordinator. self._idle = set() self._busyCount = 0 self._pending = deque() self._shouldQuitCoordinator = False self._toShrink = 0 def statistics(self): """ Gather information on the current status of this L{Team}. @return: a L{Statistics} describing the current state of this L{Team}. """ return Statistics(len(self._idle), self._busyCount, len(self._pending)) def grow(self, n): """ Increase the the number of idle workers by C{n}. @param n: The number of new idle workers to create. @type n: L{int} """ self._quit.check() @self._coordinator.do def createOneWorker(): for x in range(n): worker = self._createWorker() if worker is None: return self._recycleWorker(worker) def shrink(self, n=None): """ Decrease the number of idle workers by C{n}. @param n: The number of idle workers to shut down, or L{None} (or unspecified) to shut down all workers. @type n: L{int} or L{None} """ self._quit.check() self._coordinator.do(lambda: self._quitIdlers(n)) def _quitIdlers(self, n=None): """ The implmentation of C{shrink}, performed by the coordinator worker. @param n: see L{Team.shrink} """ if n is None: n = len(self._idle) + self._busyCount for x in range(n): if self._idle: self._idle.pop().quit() else: self._toShrink += 1 if self._shouldQuitCoordinator and self._busyCount == 0: self._coordinator.quit() def do(self, task): """ Perform some work in a worker created by C{createWorker}. @param task: the callable to run """ self._quit.check() self._coordinator.do(lambda: self._coordinateThisTask(task)) def _coordinateThisTask(self, task): """ Select a worker to dispatch to, either an idle one or a new one, and perform it. This method should run on the coordinator worker. @param task: the task to dispatch @type task: 0-argument callable """ worker = self._idle.pop() if self._idle else self._createWorker() if worker is None: # The createWorker method may return None if we're out of resources # to create workers. self._pending.append(task) return self._busyCount += 1 @worker.do def doWork(): try: task() except BaseException: self._logException() @self._coordinator.do def idleAndPending(): self._busyCount -= 1 self._recycleWorker(worker) def _recycleWorker(self, worker): """ Called only from coordinator. Recycle the given worker into the idle pool. @param worker: a worker created by C{createWorker} and now idle. @type worker: L{IWorker} """ self._idle.add(worker) if self._pending: # Re-try the first enqueued thing. # (Explicitly do _not_ honor _quit.) self._coordinateThisTask(self._pending.popleft()) elif self._shouldQuitCoordinator: self._quitIdlers() elif self._toShrink > 0: self._toShrink -= 1 self._idle.remove(worker) worker.quit() def quit(self): """ Stop doing work and shut down all idle workers. """ self._quit.set() # In case all the workers are idle when we do this. @self._coordinator.do def startFinishing(): self._shouldQuitCoordinator = True self._quitIdlers() _convenience.py 0000644 00000001576 15030470614 0007562 0 ustar 00 # -*- test-case-name: twisted._threads.test.test_convenience -*- # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Common functionality used within the implementation of various workers. """ from ._ithreads import AlreadyQuit class Quit: """ A flag representing whether a worker has been quit. @ivar isSet: Whether this flag is set. @type isSet: L{bool} """ def __init__(self): """ Create a L{Quit} un-set. """ self.isSet = False def set(self): """ Set the flag if it has not been set. @raise AlreadyQuit: If it has been set. """ self.check() self.isSet = True def check(self): """ Check if the flag has been set. @raise AlreadyQuit: If it has been set. """ if self.isSet: raise AlreadyQuit() _ithreads.py 0000644 00000003315 15030470614 0007062 0 ustar 00 # -*- test-case-name: twisted._threads.test -*- # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Interfaces related to threads. """ from zope.interface import Interface class AlreadyQuit(Exception): """ This worker worker is dead and cannot execute more instructions. """ class IWorker(Interface): """ A worker that can perform some work concurrently. All methods on this interface must be thread-safe. """ def do(task): """ Perform the given task. As an interface, this method makes no specific claims about concurrent execution. An L{IWorker}'s C{do} implementation may defer execution for later on the same thread, immediately on a different thread, or some combination of the two. It is valid for a C{do} method to schedule C{task} in such a way that it may never be executed. It is important for some implementations to provide specific properties with respect to where C{task} is executed, of course, and client code may rely on a more specific implementation of C{do} than L{IWorker}. @param task: a task to call in a thread or other concurrent context. @type task: 0-argument callable @raise AlreadyQuit: if C{quit} has been called. """ def quit(): """ Free any resources associated with this L{IWorker} and cause it to reject all future work. @raise AlreadyQuit: if this method has already been called. """ class IExclusiveWorker(IWorker): """ Like L{IWorker}, but with the additional guarantee that the callables passed to C{do} will not be called exclusively with each other. """ _threadworker.py 0000644 00000006333 15030470614 0007763 0 ustar 00 # -*- test-case-name: twisted._threads.test.test_threadworker -*- # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Implementation of an L{IWorker} based on native threads and queues. """ from zope.interface import implementer from ._convenience import Quit from ._ithreads import IExclusiveWorker _stop = object() @implementer(IExclusiveWorker) class ThreadWorker: """ An L{IExclusiveWorker} implemented based on a single thread and a queue. This worker ensures exclusivity (i.e. it is an L{IExclusiveWorker} and not an L{IWorker}) by performing all of the work passed to C{do} on the I{same} thread. """ def __init__(self, startThread, queue): """ Create a L{ThreadWorker} with a function to start a thread and a queue to use to communicate with that thread. @param startThread: a callable that takes a callable to run in another thread. @type startThread: callable taking a 0-argument callable and returning nothing. @param queue: A L{Queue} to use to give tasks to the thread created by C{startThread}. @type queue: L{Queue} """ self._q = queue self._hasQuit = Quit() def work(): for task in iter(queue.get, _stop): task() startThread(work) def do(self, task): """ Perform the given task on the thread owned by this L{ThreadWorker}. @param task: the function to call on a thread. """ self._hasQuit.check() self._q.put(task) def quit(self): """ Reject all future work and stop the thread started by C{__init__}. """ # Reject all future work. Set this _before_ enqueueing _stop, so # that no work is ever enqueued _after_ _stop. self._hasQuit.set() self._q.put(_stop) @implementer(IExclusiveWorker) class LockWorker: """ An L{IWorker} implemented based on a mutual-exclusion lock. """ def __init__(self, lock, local): """ @param lock: A mutual-exclusion lock, with C{acquire} and C{release} methods. @type lock: L{threading.Lock} @param local: Local storage. @type local: L{threading.local} """ self._quit = Quit() self._lock = lock self._local = local def do(self, work): """ Do the given work on this thread, with the mutex acquired. If this is called re-entrantly, return and wait for the outer invocation to do the work. @param work: the work to do with the lock held. """ lock = self._lock local = self._local self._quit.check() working = getattr(local, "working", None) if working is None: working = local.working = [] working.append(work) lock.acquire() try: while working: working.pop(0)() finally: lock.release() local.working = None else: working.append(work) def quit(self): """ Quit this L{LockWorker}. """ self._quit.set() self._lock = None _memory.py 0000644 00000003071 15030470614 0006566 0 ustar 00 # -*- test-case-name: twisted._threads.test.test_memory -*- # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Implementation of an in-memory worker that defers execution. """ from zope.interface import implementer from . import IWorker from ._convenience import Quit NoMoreWork = object() @implementer(IWorker) class MemoryWorker: """ An L{IWorker} that queues work for later performance. @ivar _quit: a flag indicating @type _quit: L{Quit} """ def __init__(self, pending=list): """ Create a L{MemoryWorker}. """ self._quit = Quit() self._pending = pending() def do(self, work): """ Queue some work for to perform later; see L{createMemoryWorker}. @param work: The work to perform. """ self._quit.check() self._pending.append(work) def quit(self): """ Quit this worker. """ self._quit.set() self._pending.append(NoMoreWork) def createMemoryWorker(): """ Create an L{IWorker} that does nothing but defer work, to be performed later. @return: a worker that will enqueue work to perform later, and a callable that will perform one element of that work. @rtype: 2-L{tuple} of (L{IWorker}, L{callable}) """ def perform(): if not worker._pending: return False if worker._pending[0] is NoMoreWork: return False worker._pending.pop(0)() return True worker = MemoryWorker() return (worker, perform) __init__.py 0000644 00000000771 15030470614 0006662 0 ustar 00 # -*- test-case-name: twisted.test.test_paths -*- # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Twisted integration with operating system threads. """ from ._ithreads import AlreadyQuit, IWorker from ._memory import createMemoryWorker from ._pool import pool from ._team import Team from ._threadworker import LockWorker, ThreadWorker __all__ = [ "ThreadWorker", "LockWorker", "IWorker", "AlreadyQuit", "Team", "createMemoryWorker", "pool", ]
| ver. 1.4 |
Github
|
.
| PHP 8.2.28 | Generation time: 0.02 |
proxy
|
phpinfo
|
Settings