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
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
|
"""Tests for refleaks."""
import gc
from cherrypy._cpcompat import HTTPConnection, HTTPSConnection, ntob
import threading
import cherrypy
from cherrypy import _cprequest
data = object()
def get_instances(cls):
return [x for x in gc.get_objects() if isinstance(x, cls)]
from cherrypy.test import helper
class ReferenceTests(helper.CPWebCase):
def setup_server():
class Root:
def index(self, *args, **kwargs):
cherrypy.request.thing = data
return "Hello world!"
index.exposed = True
def gc_stats(self):
output = ["Statistics:"]
# Uncollectable garbage
# gc_collect isn't perfectly synchronous, because it may
# break reference cycles that then take time to fully
# finalize. Call it twice and hope for the best.
gc.collect()
unreachable = gc.collect()
if unreachable:
output.append("\n%s unreachable objects:" % unreachable)
trash = {}
for x in gc.garbage:
trash[type(x)] = trash.get(type(x), 0) + 1
trash = [(v, k) for k, v in trash.items()]
trash.sort()
for pair in trash:
output.append(" " + repr(pair))
# Request references
reqs = get_instances(_cprequest.Request)
lenreqs = len(reqs)
if lenreqs < 2:
output.append("\nMissing Request reference. Should be 1 in "
"this request thread and 1 in the main thread.")
elif lenreqs > 2:
output.append("\nToo many Request references (%r)." % lenreqs)
for req in reqs:
output.append("Referrers for %s:" % repr(req))
for ref in gc.get_referrers(req):
if ref is not reqs:
output.append(" %s" % repr(ref))
# Response references
resps = get_instances(_cprequest.Response)
lenresps = len(resps)
if lenresps < 2:
output.append("\nMissing Response reference. Should be 1 in "
"this request thread and 1 in the main thread.")
elif lenresps > 2:
output.append("\nToo many Response references (%r)." % lenresps)
for resp in resps:
output.append("Referrers for %s:" % repr(resp))
for ref in gc.get_referrers(resp):
if ref is not resps:
output.append(" %s" % repr(ref))
return "\n".join(output)
gc_stats.exposed = True
cherrypy.tree.mount(Root())
setup_server = staticmethod(setup_server)
def test_threadlocal_garbage(self):
success = []
def getpage():
host = '%s:%s' % (self.interface(), self.PORT)
if self.scheme == 'https':
c = HTTPSConnection(host)
else:
c = HTTPConnection(host)
try:
c.putrequest('GET', '/')
c.endheaders()
response = c.getresponse()
body = response.read()
self.assertEqual(response.status, 200)
self.assertEqual(body, ntob("Hello world!"))
finally:
c.close()
success.append(True)
ITERATIONS = 25
ts = []
for _ in range(ITERATIONS):
t = threading.Thread(target=getpage)
ts.append(t)
t.start()
for t in ts:
t.join()
self.assertEqual(len(success), ITERATIONS)
self.getPage("/gc_stats")
self.assertBody("Statistics:")
|
