<s> """<STR_LIT>""" <EOL> from __future__ import division , absolute_import <EOL> import pickle , time , weakref , gc , threading <EOL> from twisted . python . compat import _PY3 <EOL> from twisted . trial import unittest <EOL> from twisted . python import threadpool , threadable , failure , context <EOL> from twisted . _threads import Team , createMemoryWorker <EOL> class Synchronization ( object ) : <EOL> failures = <NUM_LIT:0> <EOL> def __init__ ( self , N , waiting ) : <EOL> self . N = N <EOL> self . waiting = waiting <EOL> self . lock = threading . Lock ( ) <EOL> self . runs = [ ] <EOL> def run ( self ) : <EOL> if self . lock . acquire ( False ) : <EOL> if not len ( self . runs ) % <NUM_LIT:5> : <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . lock . release ( ) <EOL> else : <EOL> self . failures += <NUM_LIT:1> <EOL> self . lock . acquire ( ) <EOL> self . runs . append ( None ) <EOL> if len ( self . runs ) == self . N : <EOL> self . waiting . release ( ) <EOL> self . lock . release ( ) <EOL> synchronized = [ "<STR_LIT>" ] <EOL> threadable . synchronize ( Synchronization ) <EOL> class ThreadPoolTests ( unittest . SynchronousTestCase ) : <EOL> """<STR_LIT>""" <EOL> def getTimeout ( self ) : <EOL> """<STR_LIT>""" <EOL> return <NUM_LIT:5> <EOL> def _waitForLock ( self , lock ) : <EOL> if _PY3 : <EOL> items = range ( <NUM_LIT> ) <EOL> else : <EOL> items = xrange ( <NUM_LIT> ) <EOL> for i in items : <EOL> if lock . acquire ( False ) : <EOL> break <EOL> time . sleep ( <NUM_LIT> ) <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> def test_attributes ( self ) : <EOL> """<STR_LIT>""" <EOL> pool = threadpool . ThreadPool ( <NUM_LIT:12> , <NUM_LIT> ) <EOL> self . assertEqual ( pool . min , <NUM_LIT:12> ) <EOL> self . assertEqual ( pool . max , <NUM_LIT> ) <EOL> def test_start ( self ) : <EOL> """<STR_LIT>""" <EOL> pool = threadpool . ThreadPool ( <NUM_LIT:0> , <NUM_LIT:5> ) <EOL> pool . start ( ) <EOL> self . addCleanup ( pool . stop ) <EOL> self . assertEqual ( len ( pool . threads ) , <NUM_LIT:0> ) <EOL> pool = threadpool . ThreadPool ( <NUM_LIT:3> , <NUM_LIT:10> ) <EOL> self . assertEqual ( len ( pool . threads ) , <NUM_LIT:0> ) <EOL> pool . start ( ) <EOL> self . addCleanup ( pool . stop ) <EOL> self . assertEqual ( len ( pool . threads ) , <NUM_LIT:3> ) <EOL> def test_adjustingWhenPoolStopped ( self ) : <EOL> """<STR_LIT>""" <EOL> pool = threadpool . ThreadPool ( <NUM_LIT:0> , <NUM_LIT:5> ) <EOL> pool . start ( ) <EOL> pool . stop ( ) <EOL> pool . adjustPoolsize ( <NUM_LIT:2> ) <EOL> self . assertEqual ( len ( pool . threads ) , <NUM_LIT:0> ) <EOL> def test_threadCreationArguments ( self ) : <EOL> """<STR_LIT>""" <EOL> tp = threadpool . ThreadPool ( <NUM_LIT:0> , <NUM_LIT:1> ) <EOL> tp . start ( ) <EOL> self . addCleanup ( tp . stop ) <EOL> self . assertEqual ( tp . threads , [ ] ) <EOL> def worker ( arg ) : <EOL> pass <EOL> class Dumb ( object ) : <EOL> pass <EOL> unique = Dumb ( ) <EOL> workerRef = weakref . ref ( worker ) <EOL> uniqueRef = weakref . ref ( unique ) <EOL> tp . callInThread ( worker , unique ) <EOL> event = threading . Event ( ) <EOL> tp . callInThread ( event . set ) <EOL> event . wait ( self . getTimeout ( ) ) <EOL> del worker <EOL> del unique <EOL> gc . collect ( ) <EOL> self . assertEqual ( uniqueRef ( ) , None ) <EOL> self . assertEqual ( workerRef ( ) , None ) <EOL> def test_threadCreationArgumentsCallInThreadWithCallback ( self ) : <EOL> """<STR_LIT>""" <EOL> tp = threadpool . ThreadPool ( <NUM_LIT:0> , <NUM_LIT:1> ) <EOL> tp . start ( ) <EOL> self . addCleanup ( tp . stop ) <EOL> self . assertEqual ( tp . threads , [ ] ) <EOL> refdict = { } <EOL> onResultWait = threading . Event ( ) <EOL> onResultDone = threading . Event ( ) <EOL> resultRef = [ ] <EOL> def onResult ( success , result ) : <EOL> onResultWait . wait ( self . getTimeout ( ) ) <EOL> refdict [ '<STR_LIT>' ] = workerRef ( ) <EOL> refdict [ '<STR_LIT>' ] = uniqueRef ( ) <EOL> onResultDone . set ( ) <EOL> resultRef . append ( weakref . ref ( result ) ) <EOL> def worker ( arg , test ) : <EOL> return Dumb ( ) <EOL> class Dumb ( object ) : <EOL> pass <EOL> unique = Dumb ( ) <EOL> onResultRef = weakref . ref ( onResult ) <EOL> workerRef = weakref . ref ( worker ) <EOL> uniqueRef = weakref . ref ( unique ) <EOL> tp . callInThreadWithCallback ( onResult , worker , unique , test = unique ) <EOL> del worker <EOL> del unique <EOL> onResultWait . set ( ) <EOL> onResultDone . wait ( self . getTimeout ( ) ) <EOL> gc . collect ( ) <EOL> self . assertEqual ( uniqueRef ( ) , None ) <EOL> self . assertEqual ( workerRef ( ) , None ) <EOL> del onResult <EOL> gc . collect ( ) <EOL> self . assertEqual ( onResultRef ( ) , None ) <EOL> self . assertEqual ( resultRef [ <NUM_LIT:0> ] ( ) , None ) <EOL> def test_persistence ( self ) : <EOL> """<STR_LIT>""" <EOL> pool = threadpool . ThreadPool ( <NUM_LIT:7> , <NUM_LIT:20> ) <EOL> self . assertEqual ( pool . min , <NUM_LIT:7> ) <EOL> self . assertEqual ( pool . max , <NUM_LIT:20> ) <EOL> copy = pickle . loads ( pickle . dumps ( pool ) ) <EOL> self . assertEqual ( copy . min , <NUM_LIT:7> ) <EOL> self . assertEqual ( copy . max , <NUM_LIT:20> ) <EOL> def _threadpoolTest ( self , method ) : <EOL> """<STR_LIT>""" <EOL> N = <NUM_LIT:10> <EOL> tp = threadpool . ThreadPool ( ) <EOL> tp . start ( ) <EOL> self . addCleanup ( tp . stop ) <EOL> waiting = threading . Lock ( ) <EOL> waiting . acquire ( ) <EOL> actor = Synchronization ( N , waiting ) <EOL> for i in range ( N ) : <EOL> method ( tp , actor ) <EOL> self . _waitForLock ( waiting ) <EOL> self . assertFalse ( actor . failures , "<STR_LIT>" % <EOL> ( actor . failures , ) ) <EOL> def test_callInThread ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _threadpoolTest ( <EOL> lambda tp , actor : tp . callInThread ( actor . run ) ) <EOL> def test_callInThreadException ( self ) : <EOL> """<STR_LIT>""" <EOL> class NewError ( Exception ) : <EOL> pass <EOL> def raiseError ( ) : <EOL> raise NewError ( ) <EOL> tp = threadpool . ThreadPool ( <NUM_LIT:0> , <NUM_LIT:1> ) <EOL> tp . callInThread ( raiseError ) <EOL> tp . start ( ) <EOL> tp . stop ( ) <EOL> errors = self . flushLoggedErrors ( NewError ) <EOL> self . assertEqual ( len ( errors ) , <NUM_LIT:1> ) <EOL> def test_callInThreadWithCallback ( self ) : <EOL> """<STR_LIT>""" <EOL> waiter = threading . Lock ( ) <EOL> waiter . acquire ( ) <EOL> results = [ ] <EOL> def onResult ( success , result ) : <EOL> waiter . release ( ) <EOL> results . append ( success ) <EOL> results . append ( result ) <EOL> tp = threadpool . ThreadPool ( <NUM_LIT:0> , <NUM_LIT:1> ) <EOL> tp . callInThreadWithCallback ( onResult , lambda : "<STR_LIT:test>" ) <EOL> tp . start ( ) <EOL> try : <EOL> self . _waitForLock ( waiter ) <EOL> finally : <EOL> tp . stop ( ) <EOL> self . assertTrue ( results [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( results [ <NUM_LIT:1> ] , "<STR_LIT:test>" ) <EOL> def test_callInThreadWithCallbackExceptionInCallback ( self ) : <EOL> """<STR_LIT>""" <EOL> class NewError ( Exception ) : <EOL> pass <EOL> def raiseError ( ) : <EOL> raise NewError ( ) <EOL> waiter = threading . Lock ( ) <EOL> waiter . acquire ( ) <EOL> results = [ ] <EOL> def onResult ( success , result ) : <EOL> waiter . release ( ) <EOL> results . append ( success ) <EOL> results . append ( result ) <EOL> tp = threadpool . ThreadPool ( <NUM_LIT:0> , <NUM_LIT:1> ) <EOL> tp . callInThreadWithCallback ( onResult , raiseError ) <EOL> tp . start ( ) <EOL> try : <EOL> self . _waitForLock ( waiter ) <EOL> finally : <EOL> tp . stop ( ) <EOL> self . assertFalse ( results [ <NUM_LIT:0> ] ) <EOL> self . assertTrue ( isinstance ( results [ <NUM_LIT:1> ] , failure . Failure ) ) <EOL> self . assertTrue ( issubclass ( results [ <NUM_LIT:1> ] . type , NewError ) ) <EOL> def test_callInThreadWithCallbackExceptionInOnResult ( self ) : <EOL> """<STR_LIT>""" <EOL> class NewError ( Exception ) : <EOL> pass <EOL> waiter = threading . Lock ( ) <EOL> waiter . acquire ( ) <EOL> results = [ ] <EOL> def onResult ( success , result ) : <EOL> results . append ( success ) <EOL> results . append ( result ) <EOL> raise NewError ( ) <EOL> tp = threadpool . ThreadPool ( <NUM_LIT:0> , <NUM_LIT:1> ) <EOL> tp . callInThreadWithCallback ( onResult , lambda : None ) <EOL> tp . callInThread ( waiter . release ) <EOL> tp . start ( ) <EOL> try : <EOL> self . _waitForLock ( waiter ) <EOL> finally : <EOL> tp . stop ( ) <EOL> errors = self . flushLoggedErrors ( NewError ) <EOL> self . assertEqual ( len ( errors ) , <NUM_LIT:1> ) <EOL> self . assertTrue ( results [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( results [ <NUM_LIT:1> ] , None ) <EOL> def test_callbackThread ( self ) : <EOL> """<STR_LIT>""" <EOL> threadIds = [ ] <EOL> event = threading . Event ( ) <EOL> def onResult ( success , result ) : <EOL> threadIds . append ( threading . currentThread ( ) . ident ) <EOL> event . set ( ) <EOL> def func ( ) : <EOL> threadIds . append ( threading . currentThread ( ) . ident ) <EOL> tp = threadpool . ThreadPool ( <NUM_LIT:0> , <NUM_LIT:1> ) <EOL> tp . callInThreadWithCallback ( onResult , func ) <EOL> tp . start ( ) <EOL> self . addCleanup ( tp . stop ) <EOL> event . wait ( self . getTimeout ( ) ) <EOL> self . assertEqual ( len ( threadIds ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( threadIds [ <NUM_LIT:0> ] , threadIds [ <NUM_LIT:1> ] ) <EOL> def test_callbackContext ( self ) : <EOL> """<STR_LIT>""" <EOL> myctx = context . theContextTracker . currentContext ( ) . contexts [ - <NUM_LIT:1> ] <EOL> myctx [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> contexts = [ ] <EOL> event = threading . Event ( ) <EOL> def onResult ( success , result ) : <EOL> ctx = context . theContextTracker . currentContext ( ) . contexts [ - <NUM_LIT:1> ] <EOL> contexts . append ( ctx ) <EOL> event . set ( ) <EOL> def func ( ) : <EOL> ctx = context . theContextTracker . currentContext ( ) . contexts [ - <NUM_LIT:1> ] <EOL> contexts . append ( ctx ) <EOL> tp = threadpool . ThreadPool ( <NUM_LIT:0> , <NUM_LIT:1> ) <EOL> tp . callInThreadWithCallback ( onResult , func ) <EOL> tp . start ( ) <EOL> self . addCleanup ( tp . stop ) <EOL> event . wait ( self . getTimeout ( ) ) <EOL> self . assertEqual ( len ( contexts ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( myctx , contexts [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( myctx , contexts [ <NUM_LIT:1> ] ) <EOL> def test_existingWork ( self ) : <EOL> """<STR_LIT>""" <EOL> waiter = threading . Lock ( ) <EOL> waiter . acquire ( ) <EOL> tp = threadpool . ThreadPool ( <NUM_LIT:0> , <NUM_LIT:1> ) <EOL> tp . callInThread ( waiter . release ) <EOL> tp . start ( ) <EOL> try : <EOL> self . _waitForLock ( waiter ) <EOL> finally : <EOL> tp . stop ( ) <EOL> def test_workerStateTransition ( self ) : <EOL> """<STR_LIT>""" <EOL> pool = threadpool . ThreadPool ( <NUM_LIT:0> , <NUM_LIT:1> ) <EOL> pool . start ( ) <EOL> self . addCleanup ( pool . stop ) <EOL> self . assertEqual ( pool . workers , <NUM_LIT:0> ) <EOL> self . assertEqual ( len ( pool . waiters ) , <NUM_LIT:0> ) <EOL> self . assertEqual ( len ( pool . working ) , <NUM_LIT:0> ) <EOL> threadWorking = threading . Event ( ) <EOL> threadFinish = threading . Event ( ) <EOL> def _thread ( ) : <EOL> threadWorking . set ( ) <EOL> threadFinish . wait ( <NUM_LIT:10> ) <EOL> pool . callInThread ( _thread ) <EOL> threadWorking . wait ( <NUM_LIT:10> ) <EOL> self . assertEqual ( pool . workers , <NUM_LIT:1> ) <EOL> self . assertEqual ( len ( pool . waiters ) , <NUM_LIT:0> ) <EOL> self . assertEqual ( len ( pool . working ) , <NUM_LIT:1> ) <EOL> threadFinish . set ( ) <EOL> while not len ( pool . waiters ) : <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . assertEqual ( len ( pool . waiters ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( len ( pool . working ) , <NUM_LIT:0> ) <EOL> class RaceConditionTests ( unittest . SynchronousTestCase ) : <EOL> def setUp ( self ) : <EOL> self . threadpool = threadpool . ThreadPool ( <NUM_LIT:0> , <NUM_LIT:10> ) <EOL> self . event = threading . Event ( ) <EOL> self . threadpool . start ( ) <EOL> def done ( ) : <EOL> self . threadpool . stop ( ) <EOL> del self . threadpool <EOL> self . addCleanup ( done ) <EOL> def getTimeout ( self ) : <EOL> """<STR_LIT>""" <EOL> return <NUM_LIT:5> <EOL> def test_synchronization ( self ) : <EOL> """<STR_LIT>""" <EOL> timeout = self . getTimeout ( ) <EOL> self . threadpool . callInThread ( self . event . set ) <EOL> self . event . wait ( timeout ) <EOL> self . event . clear ( ) <EOL> for i in range ( <NUM_LIT:3> ) : <EOL> self . threadpool . callInThread ( self . event . wait ) <EOL> self . threadpool . callInThread ( self . event . set ) <EOL> self . event . wait ( timeout ) <EOL> if not self . event . isSet ( ) : <EOL> self . event . set ( ) <EOL> self . fail ( <EOL> "<STR_LIT>" <EOL> ) <EOL> def test_singleThread ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( self . threadpool . workers , <NUM_LIT:0> ) <EOL> event = threading . Event ( ) <EOL> event . clear ( ) <EOL> def onResult ( success , counter ) : <EOL> event . set ( ) <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> self . threadpool . callInThreadWithCallback ( <EOL> onResult , lambda : None ) <EOL> event . wait ( <NUM_LIT:10> ) <EOL> event . clear ( ) <EOL> self . assertEqual ( self . threadpool . workers , <NUM_LIT:1> ) <EOL> class MemoryPool ( threadpool . ThreadPool ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , coordinator , failTest , newWorker , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> self . _coordinator = coordinator <EOL> self . _failTest = failTest <EOL> self . _newWorker = newWorker <EOL> threadpool . ThreadPool . __init__ ( self , * args , ** kwargs ) <EOL> def _pool ( self , currentLimit , threadFactory ) : <EOL> """<STR_LIT>""" <EOL> def respectLimit ( ) : <EOL> stats = team . statistics ( ) <EOL> if ( stats . busyWorkerCount + stats . idleWorkerCount <EOL> >= currentLimit ( ) ) : <EOL> return None <EOL> return self . _newWorker ( ) <EOL> team = Team ( coordinator = self . _coordinator , <EOL> createWorker = respectLimit , <EOL> logException = self . _failTest ) <EOL> return team <EOL> class PoolHelper ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , testCase , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> coordinator , self . performCoordination = createMemoryWorker ( ) <EOL> self . workers = [ ] <EOL> def newWorker ( ) : <EOL> self . workers . append ( createMemoryWorker ( ) ) <EOL> return self . workers [ - <NUM_LIT:1> ] [ <NUM_LIT:0> ] <EOL> self . threadpool = MemoryPool ( coordinator , testCase . fail , newWorker , <EOL> * args , ** kwargs ) <EOL> def performAllCoordination ( self ) : <EOL> """<STR_LIT>""" <EOL> while self . performCoordination ( ) : <EOL> pass <EOL> class MemoryBackedTests ( unittest . SynchronousTestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_workBeforeStarting ( self ) : <EOL> """<STR_LIT>""" <EOL> helper = PoolHelper ( self , <NUM_LIT:0> , <NUM_LIT:10> ) <EOL> n = <NUM_LIT:5> <EOL> for x in range ( n ) : <EOL> helper . threadpool . callInThread ( lambda : None ) <EOL> helper . performAllCoordination ( ) <EOL> self . assertEqual ( helper . workers , [ ] ) <EOL> helper . threadpool . start ( ) <EOL> helper . performAllCoordination ( ) <EOL> self . assertEqual ( len ( helper . workers ) , n ) <EOL> def test_tooMuchWorkBeforeStarting ( self ) : <EOL> """<STR_LIT>""" <EOL> helper = PoolHelper ( self , <NUM_LIT:0> , <NUM_LIT:10> ) <EOL> n = <NUM_LIT:50> <EOL> for x in range ( n ) : <EOL> helper . threadpool . callInThread ( lambda : None ) <EOL> helper . performAllCoordination ( ) <EOL> self . assertEqual ( helper . workers , [ ] ) <EOL> helper . threadpool . start ( ) <EOL> helper . performAllCoordination ( ) <EOL> self . assertEqual ( len ( helper . workers ) , helper . threadpool . max ) </s>
