1import threading 2 3from TestCommon import * 4from Test import Test 5 6 7class TestThreads(Test): 8 9 def __init__(self): 10 Test.__init__(self) 11 self._numObjects = 64 12 self._numThreads = [1, 2, 4, 8, 16, 32] 13 self._numReads = 32 14 15 def timedMain(self): 16 import time 17 start = time.time() 18 self.main() 19 end = time.time() 20 duration = end - start 21 print 22 print 'secs: %0.2f' % duration 23 print 'mins: %0.2f' % (duration/60.0) 24 25 def readArgs(self, args): 26 self._modelNames = ['MKBasic'] 27 28 def testEmpty(self): 29 self.createDatabase() 30 self.createStore() 31 self.createObjects() 32 self.testConcurrentReads() 33 34 def createStore(self): 35 from MiddleKit.Run.MySQLObjectStore import MySQLObjectStore 36 self._store = MySQLObjectStore(**storeArgs) 37 self._store.readModelFileNamed(self._modelName) 38 39 def createObjects(self): 40 from Thing import Thing 41 for i in range(self._numObjects): 42 t = Thing() 43 t.setB(1) 44 t.setI(2) 45 t.setL(3) 46 t.setF(4.0) 47 t.setS('five') 48 self._store.addObject(t) 49 self._store.saveChanges() 50 things = self._store.fetchObjectsOfClass('Thing') 51 assert len(things) == self._numObjects, \ 52 '%i, %i' % (len(things), self._numObjects) 53 54 def testConcurrentReads(self): 55 for numThreads in self._numThreads: 56 print '>> numThreads:', numThreads 57 self.testReaderThreads(numThreads) 58 59 def testReaderThreads(self, numThreads): 60 61 class Reader(threading.Thread): 62 63 def __init__(self, store, numReads): 64 threading.Thread.__init__(self) 65 self._store = store 66 self._numReads = numReads 67 68 def run(self): 69 store = self._store 70 for i in range(self._numReads): 71 # print '%x:%03i' % (id(self), i), 72 objects = store.fetchObjectsOfClass('Thing') 73 74 threads = [] 75 for i in range(numThreads): 76 thread = Reader(self._store, self._numReads) 77 threads.append(thread) 78 79 for thread in threads: 80 thread.start() 81 82 for thread in threads: 83 thread.join() 84 85 def testSamples(self): 86 """Test samples. 87 88 We do all our necessary testing in testEmpty() so we override 89 this method to pass. 90 """ 91 pass 92 93 94if __name__ == '__main__': 95 import sys 96 sys.path.append('WorkDir') 97 sys.setcheckinterval(100) 98 TestThreads().timedMain() 99