def _getr(slist, olist, first=True):
i = 0
for e in slist:
oid = id(e)
typ = type(e)
if oid in olist or typ is int: ## or e in olist: ## since we're excluding all ints, there is no longer a need to check for olist keys
continue
olist[oid] = e
if first and (i%1000) == 0:
gc.collect()
tl = gc.get_referents(e)
if tl:
_getr(tl, olist, first=False)
i += 1
# The public function.
python类get_referents()的实例源码
def _getr(slist, olist, first=True):
i = 0
for e in slist:
oid = id(e)
typ = type(e)
if oid in olist or typ is int: ## or e in olist: ## since we're excluding all ints, there is no longer a need to check for olist keys
continue
olist[oid] = e
if first and (i%1000) == 0:
gc.collect()
tl = gc.get_referents(e)
if tl:
_getr(tl, olist, first=False)
i += 1
# The public function.
def memory_dump():
import gc
x = 0
for obj in gc.get_objects():
i = id(obj)
size = sys.getsizeof(obj, 0)
# referrers = [id(o) for o in gc.get_referrers(obj)]
try:
cls = str(obj.__class__)
except:
cls = "<no class>"
if size > 1024 * 50:
referents = set([id(o) for o in gc.get_referents(obj)])
x += 1
print(x, {'id': i,
'class': cls,
'size': size,
"ref": len(referents)})
#if len(referents) < 2000:
#print(obj)
def test_get_referents(self):
alist = [1, 3, 5]
got = gc.get_referents(alist)
got.sort()
self.assertEqual(got, alist)
atuple = tuple(alist)
got = gc.get_referents(atuple)
got.sort()
self.assertEqual(got, alist)
adict = {1: 3, 5: 7}
expected = [1, 3, 5, 7]
got = gc.get_referents(adict)
got.sort()
self.assertEqual(got, expected)
got = gc.get_referents([1, 2], {3: 4}, (0, 0, 0))
got.sort()
self.assertEqual(got, [0, 0] + list(range(5)))
self.assertEqual(gc.get_referents(1, 'a', 4j), [])
def __call__(self, *args):
"""Configuring calls.
"""
try:
call_info = str(gc.get_referents(self))
subst1 = "ServerProxy for "
subst2 = ">>"
proxy = call_info[call_info.find(subst1) + len(subst1):call_info.find(subst2)]
if len(args) == 0:
meth = "%s()" % self.__name
elif len(args) == 1:
if args[0] == "":
meth = "%s('')" % self.__name
else:
meth = "%s(%s)" % (self.__name, ",".join(map(str, args)))
else:
meth = "%s%s" % (self.__name, args)
if "__repr__" not in meth:
self.class_logger.debug("%s.%s" % (proxy, meth))
except Exception:
pass
return self.__send(self.__name, args)
def test_get_referents(self):
alist = [1, 3, 5]
got = gc.get_referents(alist)
got.sort()
self.assertEqual(got, alist)
atuple = tuple(alist)
got = gc.get_referents(atuple)
got.sort()
self.assertEqual(got, alist)
adict = {1: 3, 5: 7}
expected = [1, 3, 5, 7]
got = gc.get_referents(adict)
got.sort()
self.assertEqual(got, expected)
got = gc.get_referents([1, 2], {3: 4}, (0, 0, 0))
got.sort()
self.assertEqual(got, [0, 0] + range(5))
self.assertEqual(gc.get_referents(1, 'a', 4j), [])
def test_get_referents(self):
alist = [1, 3, 5]
got = gc.get_referents(alist)
got.sort()
self.assertEqual(got, alist)
atuple = tuple(alist)
got = gc.get_referents(atuple)
got.sort()
self.assertEqual(got, alist)
adict = {1: 3, 5: 7}
expected = [1, 3, 5, 7]
got = gc.get_referents(adict)
got.sort()
self.assertEqual(got, expected)
got = gc.get_referents([1, 2], {3: 4}, (0, 0, 0))
got.sort()
self.assertEqual(got, [0, 0] + range(5))
self.assertEqual(gc.get_referents(1, 'a', 4j), [])
def test_get_referents(self):
alist = [1, 3, 5]
got = gc.get_referents(alist)
got.sort()
self.assertEqual(got, alist)
atuple = tuple(alist)
got = gc.get_referents(atuple)
got.sort()
self.assertEqual(got, alist)
adict = {1: 3, 5: 7}
expected = [1, 3, 5, 7]
got = gc.get_referents(adict)
got.sort()
self.assertEqual(got, expected)
got = gc.get_referents([1, 2], {3: 4}, (0, 0, 0))
got.sort()
self.assertEqual(got, [0, 0] + list(range(5)))
self.assertEqual(gc.get_referents(1, 'a', 4j), [])
def test_get_referents(self):
alist = [1, 3, 5]
got = gc.get_referents(alist)
got.sort()
self.assertEqual(got, alist)
atuple = tuple(alist)
got = gc.get_referents(atuple)
got.sort()
self.assertEqual(got, alist)
adict = {1: 3, 5: 7}
expected = [1, 3, 5, 7]
got = gc.get_referents(adict)
got.sort()
self.assertEqual(got, expected)
got = gc.get_referents([1, 2], {3: 4}, (0, 0, 0))
got.sort()
self.assertEqual(got, [0, 0] + range(5))
self.assertEqual(gc.get_referents(1, 'a', 4j), [])
def test_get_referents(self):
alist = [1, 3, 5]
got = gc.get_referents(alist)
got.sort()
self.assertEqual(got, alist)
atuple = tuple(alist)
got = gc.get_referents(atuple)
got.sort()
self.assertEqual(got, alist)
adict = {1: 3, 5: 7}
expected = [1, 3, 5, 7]
got = gc.get_referents(adict)
got.sort()
self.assertEqual(got, expected)
got = gc.get_referents([1, 2], {3: 4}, (0, 0, 0))
got.sort()
self.assertEqual(got, [0, 0] + list(range(5)))
self.assertEqual(gc.get_referents(1, 'a', 4j), [])
def test_get_referents(self):
alist = [1, 3, 5]
got = gc.get_referents(alist)
got.sort()
self.assertEqual(got, alist)
atuple = tuple(alist)
got = gc.get_referents(atuple)
got.sort()
self.assertEqual(got, alist)
adict = {1: 3, 5: 7}
expected = [1, 3, 5, 7]
got = gc.get_referents(adict)
got.sort()
self.assertEqual(got, expected)
got = gc.get_referents([1, 2], {3: 4}, (0, 0, 0))
got.sort()
self.assertEqual(got, [0, 0] + range(5))
self.assertEqual(gc.get_referents(1, 'a', 4j), [])
def test_get_referents(self):
alist = [1, 3, 5]
got = gc.get_referents(alist)
got.sort()
self.assertEqual(got, alist)
atuple = tuple(alist)
got = gc.get_referents(atuple)
got.sort()
self.assertEqual(got, alist)
adict = {1: 3, 5: 7}
expected = [1, 3, 5, 7]
got = gc.get_referents(adict)
got.sort()
self.assertEqual(got, expected)
got = gc.get_referents([1, 2], {3: 4}, (0, 0, 0))
got.sort()
self.assertEqual(got, [0, 0] + list(range(5)))
self.assertEqual(gc.get_referents(1, 'a', 4j), [])
def _getreferents(unused):
return () # sorry, no refs
# sys.getsizeof() new in Python 2.6, adjusted below
def _iter_refs(obj, named):
'''Return the referent(s) of an iterator object.
'''
r = _getreferents(obj) # special case
return _refs(r, named, itor=_nameof(obj) or 'iteref')
def _getreferents(unused):
return () # sorry, no refs
# sys.getsizeof() new in Python 2.6
def _iter_refs(obj, named):
'''Return the referent(s) of an iterator object.
'''
r = _getreferents(obj) # special case
return _refs(r, named, itor=_nameof(obj) or 'iteref')
def inspectGeneral(self, current):
for ii, object in enumerate(gc.get_referents(current.object)):
if type(object) is type:
continue
yield self.Record(object, '%s/%d' % (current.path, ii))
def _memory_dump(opts):
for typ, n in objgraph.most_common_types():
logging.info('{typ:30} {n:>10}'.format(typ=typ, n=n))
objects = []
rng = opts['size_range']
summ = {
'max_refsize': {
'size': 0,
},
}
for obj in gc.get_objects():
if not hasattr(obj, '__class__'):
continue
size = sys.getsizeof(obj, 0)
if rng is not None:
if not (rng[0] <= size < rng[1]):
continue
i = id(obj)
# referrers = [id(o)
# for o in gc.get_referrers(obj)
# if hasattr(o, '__class__')]
referents = [(id(o), _get_class(o), sys.getsizeof(o, 0))
for o in gc.get_referents(obj)
# if hasattr(o, '__class__')
]
refsize = sum([x[2] for x in referents])
cls = _get_class(obj)
data = [
i,
cls,
size, # object size
refsize, # size of all direct referents
referents, # referents
]
objects.append(data)
if summ['max_refsize']['size'] < refsize:
summ['max_refsize'] = {
'size': refsize,
'object': data,
}
for o in objects:
logging.info('memory-dump: ' + json.dumps(o))
logging.info('memory-dump summary: ' + json.dumps(summ))
def memory_dump(show_cycles=True, show_objects=False):
print "\nGARBAGE:"
gc.collect()
garbage = gc.garbage[:]
if show_cycles:
nodes = {id(obj): Node(obj) for obj in garbage}
for obj in garbage:
nodes[id(obj)].successors = tuple(nodes[id(s)] for s in gc.get_referents(obj) if id(s) in nodes)
nodes[id(obj)].visitable_successors = deque(nodes[id(obj)].successors)
cycles = set()
remaining_nodes = nodes.copy()
while remaining_nodes:
path = [next(remaining_nodes.itervalues())]
while path:
node = path[-1]
remaining_nodes.pop(id(node.object), None)
if node.visitable_successors:
successor = node.visitable_successors.pop()
if successor in path:
cycles.add(Cycle(n.object for n in path[path.index(successor):]))
else:
path.append(successor)
else:
node.visitable_successors = deque(node.successors)
path.pop(-1)
for node in nodes.itervalues():
node.successors = node.visitable_successors = None
print "\nCOLLECTABLE CYCLES:"
for cycle in (c for c in cycles if c.collectable):
print cycle
print "\nUNCOLLECTABLE CYCLES:"
for cycle in (c for c in cycles if not c.collectable):
print cycle
if show_objects:
try:
import fcntl, struct, sys, termios
console_width = struct.unpack('HHHH', fcntl.ioctl(sys.stdout.fileno(), termios.TIOCGWINSZ, struct.pack('HHHH', 0, 0, 0, 0)))[1]
except Exception:
console_width = 80
print "\nGARBAGE OBJECTS:"
for x in garbage:
s = str(x)
if len(s) > console_width-2:
s = s[:console_width-5] + '...'
print "%s\n %s" % (type(x), s)
