Repository navigation
Expand file tree
/
Copy pathstress.py
More file actions
77 lines (61 loc) · 2.03 KB
/
Copy pathstress.py
File metadata and controls
77 lines (61 loc) · 2.03 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# FIXME: FAIL: testImplicitAssemblyLoad AssertionError: 0 != 1
"""
Run all of the unit tests for this package multiple times in a highly
multi-threaded way to stress the system. This makes it possible to look
for memory leaks and threading issues and provides a good target for a
profiler to accumulate better data.
"""
from __future__ import print_function
import gc
import os
import sys
import threading
import time
import _thread as thread
from .utils import dprint
class StressTest(object):
def __init__(self):
self.dirname = os.path.split(__file__)[0]
sys.path.append(self.dirname)
gc.set_debug(gc.DEBUG_LEAK)
import runtests
self.module = runtests
self.done = []
def mark_start(self):
self._start = time.clock()
def mark_finish(self):
self._finish = time.clock()
def elapsed(self):
return self._finish - self._start
def print_gc_report(self):
for item in gc.get_objects():
print(item, sys.getrefcount(item))
def run_thread(self, iterations):
thread_id = thread.get_ident()
dprint("thread {0} starting...".format(thread_id))
time.sleep(0.1)
for i in range(iterations):
dprint("thread {0} iter {1} start".format(thread_id, i))
self.module.main()
dprint("thread {0} iter {1} end".format(thread_id, i))
self.done.append(None)
dprint("thread {0} done".format(thread_id))
def stress_test(self, iterations=1, threads=1):
args = (iterations,)
self.mark_start()
for _ in range(threads):
thread = threading.Thread(target=self.run_thread, args=args)
thread.start()
while len(self.done) < (iterations * threads):
dprint(len(self.done))
time.sleep(0.1)
self.mark_finish()
took = self.elapsed()
self.print_gc_report()
def main():
test = StressTest()
test.stress_test(2, 10)
if __name__ == '__main__':
main()