Files
controller-wisetopdataserver/python/biopro/util/leak.py
T
2021-12-20 14:52:55 +08:00

105 lines
2.4 KiB
Python

import abc
import sys
from typing import IO, Union
class LeakSection(metaclass=abc.ABCMeta):
__slots__ = ()
@abc.abstractmethod
def reset(self):
pass
@abc.abstractmethod
def check_point(self):
pass
@abc.abstractmethod
def dump(self, types, file: Union[None, str, IO] = None):
pass
def __enter__(self):
self.reset()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.check_point()
class NoLeakSection(LeakSection):
__slots__ = ()
def reset(self):
pass
def check_point(self):
pass
def dump(self, types, file: Union[None, str, IO] = None):
pass
class ObjGraphLeakSection(LeakSection):
__slots__ = ('objgraph',)
def __init__(self):
import objgraph
self.objgraph = objgraph
def reset(self):
self.objgraph.growth()
def check_point(self):
self.objgraph.show_growth(3)
def dump(self, types, file: Union[None, str, IO] = None):
if file is None:
self._dump(types, sys.stdout)
elif isinstance(file, str):
with open(file, 'w') as f:
self._dump(types, f)
else:
self._dump(types, file)
def _dump(self, types, file):
roots = self.objgraph.get_leaking_objects()
for i, leak in enumerate(filter(self._leak_test(types), roots)):
print('%d)' % i, '+%d' % sys.getrefcount(leak), leak, file=file)
@staticmethod
def _leak_test(types):
def _inner(obj):
if isinstance(obj, list) and list in types:
return True
elif isinstance(obj, dict) and dict in types:
if '__repr__' in obj or \
'__iter__' in obj or \
'__init__' in obj or \
'__new__' in obj or \
'__name__' in obj or \
'__hash__' in obj or \
'__doc__' in obj:
return False
for obj_value in obj.values():
if 'weakref' in repr(obj_value):
return False
return True
elif isinstance(obj, tuple) and tuple in types:
return True
else:
for t in types:
if isinstance(obj, t):
return True
return False
return _inner