import sys ############# hash_values = [3, 2, 5, 6, 5, 1, 0, 3, 4, 2, 0, 1, 3, 2, 3, 4, 7, 7, 7, 2, 0, 1, 4, 6, 0, 3, 3, 1, 3, 2, 0, 7, 7, 7, 2] class Thread2: __slots__ = ('hashvalue',) def __init__(self, hashvalue): self.hashvalue = hashvalue def __hash__(self): return self.hashvalue items = [Thread2(hash_values[i % len(hash_values)]) for i in range(11)] dangling = set() counter = 1 def test_current_frames(): global counter, dangling i = counter % len(items) counter += 1 dangling.add(items[0]) dangling.add(items[i]) dangling.discard(items[i]) dangling.discard(items[0]) assert len(dangling) == 0 ############### def dash_R(): nwarmup, ntracked = (4, 30) repcount = nwarmup + ntracked alloc_deltas = [0] * ntracked # prealloc all integers for sys.getallocatedblocks() result integers = list(range(20_000)) alloc_before = 0 alloc_after = 0 for i in range(nwarmup): test_current_frames() alloc_before = sys.getallocatedblocks() # The leak occurs somewhere here for i in range(ntracked): test_current_frames() if 0: print(sys.getallocatedblocks()) else: alloc_after = sys.getallocatedblocks() alloc_deltas[i] = alloc_after - alloc_before alloc_before = alloc_after i = None # --- if any(delta >= 1 for delta in alloc_deltas): msg = 'test_leak leaked %s %s, sum=%s' % ( alloc_deltas, 'memory blocks', sum(alloc_deltas)) print(msg, file=sys.stderr, flush=True) failed = True else: print("no leak") failed = False return failed def warm_caches(): # char cache s = bytes(range(256)) for i in range(256): s[i:i+1] # unicode cache [chr(i) for i in range(256)] # int cache list(range(-5, 257)) if __name__ == "__main__": warm_caches() dash_R()