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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
|
import shutil
import subprocess
import sys
from pathlib import Path
from textwrap import dedent
import pytest
from memray import AllocatorType
from memray import FileReader
from memray import Tracker
from memray._test import MemoryAllocator
HERE = Path(__file__).parent
TEST_MULTITHREADED_EXTENSION = HERE / "multithreaded_extension"
TEST_MISBEHAVING_EXTENSION = HERE / "misbehaving_extension"
TEST_RPATH_EXTENSION = HERE / "rpath_extension"
@pytest.mark.valgrind
def test_multithreaded_extension(tmpdir, monkeypatch):
"""Test tracking allocations in a native extension which spawns multiple threads,
each thread allocating and freeing memory."""
# GIVEN
output = Path(tmpdir) / "test.bin"
extension_name = "multithreaded_extension"
extension_path = tmpdir / extension_name
shutil.copytree(TEST_MULTITHREADED_EXTENSION, extension_path)
subprocess.run(
[sys.executable, str(extension_path / "setup.py"), "build_ext", "--inplace"],
check=True,
cwd=extension_path,
capture_output=True,
)
# WHEN
with monkeypatch.context() as ctx:
ctx.setattr(sys, "path", [*sys.path, str(extension_path)])
from testext import run # type: ignore
with Tracker(output):
run()
# THEN
records = list(FileReader(output).get_allocation_records())
assert records
memaligns = [
record for record in records if record.allocator == AllocatorType.POSIX_MEMALIGN
]
assert len(memaligns) == 100 * 100 # 100 threads allocate 100 times in testext
# We don't keep track of the native stacks. Make sure they are empty
assert all(len(memalign.stack_trace()) == 0 for memalign in memaligns)
memaligns_addr = {record.address for record in memaligns}
memalign_frees = [
record
for record in records
if record.address in memaligns_addr and record.allocator == AllocatorType.FREE
]
assert len(memalign_frees) >= 100 * 100
def test_misbehaving_extension(tmpdir, monkeypatch):
"""Check that we can correctly track allocations in an extension which invokes
Python code in a thread and does not register trace functions."""
# GIVEN
output = Path(tmpdir) / "test.bin"
extension_name = "misbehaving_extension"
extension_path = tmpdir / extension_name
shutil.copytree(TEST_MISBEHAVING_EXTENSION, extension_path)
subprocess.run(
[sys.executable, str(extension_path / "setup.py"), "build_ext", "--inplace"],
check=True,
cwd=extension_path,
capture_output=True,
)
def allocating_function(): # pragma: no cover
allocator = MemoryAllocator()
allocator.valloc(1234)
allocator.free()
# WHEN
with monkeypatch.context() as ctx:
ctx.setattr(sys, "path", [*sys.path, str(extension_path)])
from misbehaving import call_fn # type: ignore
with Tracker(output):
call_fn(allocating_function)
# THEN
allocations = list(FileReader(output).get_allocation_records())
allocs = [
event
for event in allocations
if event.size == 1234 and event.allocator == AllocatorType.VALLOC
]
assert len(allocs) == 1
(alloc,) = allocs
stack_trace = alloc.stack_trace()
assert len(stack_trace)
*_, bottom_frame = stack_trace
func, filename, line = bottom_frame
assert func == "allocating_function"
assert filename.endswith(__file__)
assert line == 83
frees = [
event
for event in allocations
if event.address == alloc.address and event.allocator == AllocatorType.FREE
]
assert len(frees) >= 1
def test_extension_that_uses_pygilstate_ensure(tmpdir, monkeypatch):
"""Check that we can correctly track allocations in an extension which invokes
Python code in a thread and does not register trace functions."""
# GIVEN
output = Path(tmpdir) / "test.bin"
extension_name = "misbehaving_extension"
extension_path = tmpdir / extension_name
shutil.copytree(TEST_MISBEHAVING_EXTENSION, extension_path)
subprocess.run(
[sys.executable, str(extension_path / "setup.py"), "build_ext", "--inplace"],
check=True,
cwd=extension_path,
capture_output=True,
)
def allocating_function():
allocator = MemoryAllocator()
allocator.valloc(1234)
allocator.free()
def foo1():
foo2()
def foo2():
call_fn_no_thread(allocating_function)
# WHEN
with monkeypatch.context() as ctx:
ctx.setattr(sys, "path", [*sys.path, str(extension_path)])
from misbehaving import call_fn_no_thread
allocator = MemoryAllocator()
with Tracker(output):
foo1()
allocator.valloc(1234)
allocator.free()
# THEN
allocations = list(FileReader(output).get_allocation_records())
allocs = [
event
for event in allocations
if event.size == 1234 and event.allocator == AllocatorType.VALLOC
]
assert len(allocs) == 2
(alloc1, alloc2) = allocs
stack_trace = alloc1.stack_trace()
assert len(stack_trace)
first_frame, *_, bottom_frame = stack_trace
func, filename, line = bottom_frame
assert func == "test_extension_that_uses_pygilstate_ensure"
assert filename.endswith(__file__)
assert line == 154
# We should have 2 frames here: this function calling `allocator.valloc`,
# and `allocator.valloc` calling the C `valloc`.
# We should not see any call to foo1() or foo2().
stack_trace = alloc2.stack_trace()
assert len(stack_trace) == 2
(callee, caller) = stack_trace
func, filename, line = callee
assert func == "valloc"
assert filename.endswith("/_test.py")
func, filename, line = caller
assert func == "test_extension_that_uses_pygilstate_ensure"
assert filename.endswith(__file__)
assert line == 155
frees = [
event
for event in allocations
if event.address in (alloc1.address, alloc2.address)
and event.allocator == AllocatorType.FREE
]
assert len(frees) >= 1
def test_native_dlopen(tmpdir, monkeypatch):
"""Check that we can correctly track allocations in an extension which calls
dlopen() without the GIL held"""
# GIVEN
output = Path(tmpdir) / "test.bin"
extension_name = "misbehaving_extension"
extension_path = tmpdir / extension_name
shutil.copytree(TEST_MISBEHAVING_EXTENSION, extension_path)
subprocess.run(
[sys.executable, str(extension_path / "setup.py"), "build_ext", "--inplace"],
check=True,
cwd=extension_path,
capture_output=True,
)
def allocating_function():
allocator = MemoryAllocator()
allocator.valloc(1234)
allocator.free()
# WHEN
with monkeypatch.context() as ctx:
ctx.setattr(sys, "path", [*sys.path, str(extension_path)])
from misbehaving import dlopen_self # type: ignore
with Tracker(output):
dlopen_self(allocating_function)
# THEN
allocations = list(FileReader(output).get_allocation_records())
allocs = [
event
for event in allocations
if event.size == 1234 and event.allocator == AllocatorType.VALLOC
]
assert len(allocs) == 1
(alloc,) = allocs
stack_trace = alloc.stack_trace()
assert len(stack_trace)
*_, bottom_frame = stack_trace
func, filename, line = bottom_frame
assert func == "test_native_dlopen"
assert filename.endswith(__file__)
assert line == 226
frees = [
event
for event in allocations
if event.address == alloc.address and event.allocator == AllocatorType.FREE
]
assert len(frees) >= 1
@pytest.mark.valgrind
def test_valloc_at_thread_exit(tmpdir, monkeypatch):
"""Test tracking allocations that happen while a thread is shutting down"""
# GIVEN
output = Path(tmpdir) / "test.bin"
extension_name = "multithreaded_extension"
extension_path = tmpdir / extension_name
shutil.copytree(TEST_MULTITHREADED_EXTENSION, extension_path)
subprocess.run(
[sys.executable, str(extension_path / "setup.py"), "build_ext", "--inplace"],
check=True,
cwd=extension_path,
capture_output=True,
)
# WHEN
with monkeypatch.context() as ctx:
ctx.setattr(sys, "path", [*sys.path, str(extension_path)])
from testext import run_valloc_at_exit # type: ignore
with Tracker(output, native_traces=True):
run_valloc_at_exit()
# THEN
records = list(FileReader(output).get_allocation_records())
assert records
vallocs = [record for record in records if record.allocator == AllocatorType.VALLOC]
assert len(vallocs) == 1
def test_valloc_at_thread_exit_in_subprocess(tmpdir, monkeypatch):
"""Test tracking allocations in the destructor of a TLS variable.
Ensure that TLS variable is created before Memray is imported.
"""
# GIVEN
output = Path(tmpdir) / "test.bin"
extension_name = "multithreaded_extension"
extension_path = tmpdir / extension_name
shutil.copytree(TEST_MULTITHREADED_EXTENSION, extension_path)
subprocess.run(
[sys.executable, str(extension_path / "setup.py"), "build_ext", "--inplace"],
check=True,
cwd=extension_path,
capture_output=True,
)
code = dedent(
f"""
from testext import run_valloc_at_exit
run_valloc_at_exit() # First call creates the test extension TLS.
from memray import Tracker
with Tracker({str(output)!r}, native_traces=True):
run_valloc_at_exit()
"""
)
# WHEN
with monkeypatch.context() as ctx:
ctx.setenv("PYTHONPATH", str(extension_path), prepend=":")
subprocess.run(
[sys.executable, "-c", code],
check=True,
)
# THEN
records = list(FileReader(output).get_allocation_records())
assert records
vallocs = [record for record in records if record.allocator == AllocatorType.VALLOC]
assert len(vallocs) == 1
@pytest.mark.parametrize("py_finalize", [True, False])
def test_hard_exit(tmpdir, py_finalize):
"""Test a program that exits directly under the context manager"""
# GIVEN
# Run a program that calls from memray._test.exit under a Tracker context
# manager and check that it finishes without error
output = Path(tmpdir) / "test.bin"
code = dedent(
f"""\
from memray._test import exit
from memray import Tracker
with Tracker("{output}"):
exit(py_finalize={py_finalize})
"""
)
# WHEN
subprocess.run(
[sys.executable, "-c", code],
check=True,
capture_output=True,
)
# THEN
# No assertions, just check that the program exits without error
@pytest.mark.skipif(
sys.platform == "darwin", reason="Test requires a linker that supports $ORIGIN"
)
def test_dlopen_with_rpath(tmpdir, monkeypatch):
# GIVEN
output = Path(tmpdir) / "test.bin"
extension_name = "sharedlibs"
extension_path = tmpdir / extension_name
shutil.copytree(TEST_RPATH_EXTENSION, extension_path)
subprocess.run(
[sys.executable, str(extension_path / "setup.py"), "build_ext", "--inplace"],
check=True,
cwd=extension_path,
capture_output=True,
)
# WHEN
with monkeypatch.context() as ctx:
ctx.setattr(sys, "path", [*sys.path, str(extension_path)])
from ext import hello_world # type: ignore
try:
hello_world()
except RuntimeError:
pytest.skip("Test requires a linker that supports -rpath")
# THEN
with Tracker(output):
hello_world()
|
