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