rc_before = alloc_before = fd_before = interned_before = 0
if not quiet:
- print("beginning", repcount, "repetitions", file=sys.stderr)
- print(("1234567890"*(repcount//10 + 1))[:repcount], file=sys.stderr,
- flush=True)
+ print("beginning", repcount, "repetitions. Showing number of leaks "
+ "(. for 0 or less, X for 10 or more)",
+ file=sys.stderr)
+ numbers = ("1234567890"*(repcount//10 + 1))[:repcount]
+ numbers = numbers[:warmups] + ':' + numbers[warmups:]
+ print(numbers, file=sys.stderr, flush=True)
results = None
dash_R_cleanup(fs, ps, pic, zdc, abcs)
rc_after = gettotalrefcount() - interned_after * 2
fd_after = fd_count()
- if not quiet:
- print('.', end='', file=sys.stderr, flush=True)
-
rc_deltas[i] = get_pooled_int(rc_after - rc_before)
alloc_deltas[i] = get_pooled_int(alloc_after - alloc_before)
fd_deltas[i] = get_pooled_int(fd_after - fd_before)
+ if not quiet:
+ # use max, not sum, so total_leaks is one of the pooled ints
+ total_leaks = max(rc_deltas[i], alloc_deltas[i], fd_deltas[i])
+ if total_leaks <= 0:
+ symbol = '.'
+ elif total_leaks < 10:
+ symbol = (
+ '.', '1', '2', '3', '4', '5', '6', '7', '8', '9',
+ )[total_leaks]
+ else:
+ symbol = 'X'
+ if i == warmups:
+ print(' ', end='', file=sys.stderr, flush=True)
+ print(symbol, end='', file=sys.stderr, flush=True)
+ del total_leaks
+ del symbol
+
alloc_before = alloc_after
rc_before = rc_after
fd_before = fd_after
]:
# ignore warmup runs
deltas = deltas[warmups:]
- if checker(deltas):
+ failing = checker(deltas)
+ suspicious = any(deltas)
+ if failing or suspicious:
msg = '%s leaked %s %s, sum=%s' % (
test_name, deltas, item_name, sum(deltas))
- print(msg, file=sys.stderr, flush=True)
- with open(filename, "a", encoding="utf-8") as refrep:
- print(msg, file=refrep)
- refrep.flush()
- failed = True
+ print(msg, end='', file=sys.stderr)
+ if failing:
+ print(file=sys.stderr, flush=True)
+ with open(filename, "a", encoding="utf-8") as refrep:
+ print(msg, file=refrep)
+ refrep.flush()
+ failed = True
+ else:
+ print(' (this is fine)', file=sys.stderr, flush=True)
return (failed, results)
stderr=subprocess.STDOUT)
self.check_executed_tests(output, [test], failed=test, stats=1)
- line = 'beginning 6 repetitions\n123456\n......\n'
- self.check_line(output, re.escape(line))
+ line = r'beginning 6 repetitions. .*\n123:456\n[.0-9X]{3} 111\n'
+ self.check_line(output, line)
line2 = '%s leaked [1, 1, 1] %s, sum=3\n' % (test, what)
self.assertIn(line2, output)