leak.py

import gc
import sys
import time
import itertools
import weakref
import concurrent.futures
import tracemalloc
import ctypes


class MallInfo(ctypes.Structure):
    _fields_ = [(name, ctypes.c_int)
                for name in ('arena', 'ordblks', 'smblks', 'hblks', 'hblkhd',
                             'usmblks', 'fsmblks', 'uordblks', 'fordblks',
                             'keepcost')]

libc = ctypes.CDLL("libc.so.6")
mallinfo = libc.mallinfo
mallinfo.argtypes = []
mallinfo.restype = MallInfo

tracemalloc.start()


class ByteArray(bytearray):
    _instance_iter = itertools.count()
    instances = weakref.WeakValueDictionary()

    def __new__(cls, *args, **kwargs):
        self = bytearray.__new__(cls, *args, **kwargs)
        cls.instances[next(cls._instance_iter)] = self
        return self


def yoba(a):
    #pass
    a = ByteArray(a)


def show_mallinfo():
    info = mallinfo()
    fields = [(name, getattr(info, name)) for name, _ in info._fields_]
    print("Malloc info:")
    for name, value in fields:
        print(f"- {name}: {value}")


def show_instances():
    print("%d ByteArray instances" % len(ByteArray.instances))


show_mallinfo()


sizer = itertools.cycle(range(2, 10))
with concurrent.futures.ThreadPoolExecutor(max_workers=12) as e:
    for i in range(200):
        size = 2 ** 20 * next(sizer)
        #print(i, size)
        e.submit(yoba, ByteArray([1, 2, 3, 4]) * size).result()
        #e.submit(yoba, ByteArray([1, 2, 3, 4]) * size)
        #show_instances()


print("done")
show_instances()

for i in range(3):
    gc.collect()
    time.sleep(0.1)

show_instances()

show_mallinfo()
#sys._debugmallocstats()