memfinder.py

"""
Memory Finder

- Use case: Find out what's holding on to a flow to troubleshoot garbage collection
- Holds a reference to the first flow in a capture
- Prints out who else is holding on to the flow every 5 seconds
"""
import threading
import weakref
import gc
import time
import inspect
import collections

first_flow = lambda: None

def pprint(x, indent=""):
	print(indent + type(x).__name__ + "@" + str(id(x)) + ": " + str(x).replace("\r\n",";")[:60])

def print_refs(x, depth, parent=None, indent = ""):
	gc.collect()
	refs = tuple(sorted(gc.get_referrers(x), key=id))
	for ref in refs:
		if ref in (parent, inspect.currentframe()):
			continue
		pprint(ref, indent)
		print(indent + "-" * 10)
		if depth:
			print_refs(ref, depth-1, refs, indent + " " * 4)

def print_state():
	cframe = inspect.currentframe()
	while True:
		gc.collect()
		if False:
			print("#" * 10)
			f = first_flow()
			if f:
				print(">" * 10)
				print_refs(f, 2, cframe)
			f = None

		print("*" * 10)

		objects = gc.get_objects()
		objs = collections.Counter(type(x) for x in objects)
		print(len(objects))
		print(",\n".join(str(x) for x in objs.most_common(80)))
		objects = None
		objs = None

		time.sleep(1)
		

class MemFinder:

	done: bool = False

	def load(self, loader):
		print("start")
		threading.Thread(target=print_state, daemon=True).start()

	def request(self, flow):
		"""
		Called when a server response has been received.
		"""
		if not self.done:
			global first_flow
			first_flow = weakref.ref(flow)
			self.done = True

addons = [
	MemFinder()
]