OLD | NEW |
| (Empty) |
1 """Measure latency of reactor thread APIs. run with runtests.""" | |
2 | |
3 from pyunit import unittest | |
4 import time | |
5 | |
6 from twisted.internet import reactor, threads | |
7 | |
8 | |
9 class LatencyTestCase(unittest.TestCase): | |
10 | |
11 numRounds = 5 | |
12 | |
13 def setUp(self): | |
14 self.from_times = [] | |
15 self.in_times = [] | |
16 | |
17 def tearDown(self): | |
18 threads.shutdown() | |
19 | |
20 def wait(self): | |
21 start = time.time() | |
22 while time.time() - start < 1: | |
23 reactor.iterate(1.0) | |
24 | |
25 def printResult(self): | |
26 print | |
27 print | |
28 print "callFromThread latency:" | |
29 sum = 0 | |
30 for t in self.from_times: sum += t | |
31 print "%f millisecond" % ((sum / self.numRounds) * 1000) | |
32 | |
33 print "callInThread latency:" | |
34 sum = 0 | |
35 for t in self.in_times: sum += t | |
36 print "%f millisecond" % ((sum / self.numRounds) * 1000) | |
37 print | |
38 print | |
39 | |
40 def testCallFromThread(self): | |
41 for i in range(self.numRounds): | |
42 reactor.callInThread(self.tcmf_2, time.time()) | |
43 self.wait() | |
44 assert len(self.in_times) == len(self.from_times) | |
45 assert len(self.in_times) == self.numRounds | |
46 self.printResult() | |
47 | |
48 def tcmf_2(self, start): | |
49 # runs in thread | |
50 self.in_times.append(time.time() - start) | |
51 reactor.callFromThread(self.tcmf_3, time.time()) | |
52 | |
53 def tcmf_3(self, start): | |
54 self.from_times.append(time.time() - start) | |
OLD | NEW |