1#!/usr/bin/env python3 2 3from unittest import main, skipUnless, TestCase 4from utils import kernel_version_ge 5import os 6import subprocess 7import sys 8import tempfile 9 10TOOLS_DIR = "/bcc/tools/" 11 12 13class cfg: 14 cmd_format = "" 15 16 # Amount of memory to leak. Note, that test application allocates memory 17 # for its own needs in libc, so this amount should be large enough to be 18 # the biggest allocation. 19 leaking_amount = 30000 20 21 22def setUpModule(): 23 # Build the memory leaking application. 24 c_src = 'test_tools_memleak_leaker_app.c' 25 tmp_dir = tempfile.mkdtemp(prefix='bcc-test-memleak-') 26 c_src_full = os.path.dirname(sys.argv[0]) + os.path.sep + c_src 27 exec_dst = tmp_dir + os.path.sep + 'leaker_app' 28 29 if subprocess.call(['gcc', '-g', '-O0', '-o', exec_dst, c_src_full]) != 0: 30 print("can't compile the leaking application") 31 raise Exception 32 33 # Taking two snapshot with one second interval. Getting the largest 34 # allocation. Since attaching to a program happens with a delay, we wait 35 # for the first snapshot, then issue the command to the app. Finally, 36 # second snapshot is used to extract the information. 37 # Helper utilities "timeout" and "setbuf" are used to limit overall running 38 # time, and to disable buffering. 39 cfg.cmd_format = ( 40 'stdbuf -o 0 -i 0 timeout -s KILL 10s ' + TOOLS_DIR + 41 'memleak.py -c "{} {{}} {}" -T 1 1 2'.format(exec_dst, 42 cfg.leaking_amount)) 43 44 45@skipUnless(kernel_version_ge(4, 6), "requires kernel >= 4.6") 46class MemleakToolTests(TestCase): 47 def tearDown(self): 48 if self.p: 49 del(self.p) 50 def run_leaker(self, leak_kind): 51 # Starting memleak.py, which in turn launches the leaking application. 52 self.p = subprocess.Popen(cfg.cmd_format.format(leak_kind), 53 stdin=subprocess.PIPE, stdout=subprocess.PIPE, 54 shell=True) 55 56 # Waiting for the first report. 57 while True: 58 self.p.poll() 59 if self.p.returncode is not None: 60 break 61 line = self.p.stdout.readline() 62 if b"with outstanding allocations" in line: 63 break 64 65 # At this point, memleak.py have already launched application and set 66 # probes. Sending command to the leaking application to make its 67 # allocations. 68 out = self.p.communicate(input=b"\n")[0] 69 70 # If there were memory leaks, they are in the output. Filter the lines 71 # containing "byte" substring. Every interesting line is expected to 72 # start with "N bytes from" 73 x = [x for x in out.split(b'\n') if b'byte' in x] 74 75 self.assertTrue(len(x) >= 1, 76 msg="At least one line should have 'byte' substring.") 77 78 # Taking last report. 79 x = x[-1].split() 80 self.assertTrue(len(x) >= 1, 81 msg="There should be at least one word in the line.") 82 83 # First word is the leak amount in bytes. 84 return int(x[0]) 85 86 def test_malloc(self): 87 self.assertEqual(cfg.leaking_amount, self.run_leaker("malloc")) 88 89 def test_calloc(self): 90 self.assertEqual(cfg.leaking_amount, self.run_leaker("calloc")) 91 92 def test_realloc(self): 93 self.assertEqual(cfg.leaking_amount, self.run_leaker("realloc")) 94 95 def test_posix_memalign(self): 96 self.assertEqual(cfg.leaking_amount, self.run_leaker("posix_memalign")) 97 98 def test_valloc(self): 99 self.assertEqual(cfg.leaking_amount, self.run_leaker("valloc")) 100 101 def test_memalign(self): 102 self.assertEqual(cfg.leaking_amount, self.run_leaker("memalign")) 103 104 def test_pvalloc(self): 105 self.assertEqual(cfg.leaking_amount, self.run_leaker("pvalloc")) 106 107 def test_aligned_alloc(self): 108 self.assertEqual(cfg.leaking_amount, self.run_leaker("aligned_alloc")) 109 110 111if __name__ == "__main__": 112 main() 113