我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用gc.garbage()。
def test_cleanup(self): gc.enable() gc.collect() assert not gc.garbage, "Object leak: %s" % str(gc.garbage) container = pyngus.Container("abc") c1 = container.create_connection("c1") c2 = container.create_connection("c2") assert c2 del c2 gc.collect() c2 = container.get_connection("c2") assert c2 c1 = container.get_connection("c1") assert c1 c1.create_receiver("r1") c1.create_sender("s1") del c1 del c2 container.destroy() del container gc.collect() assert not gc.garbage, "Object leak: %s" % str(gc.garbage)
def scan_and_print_new_objs(self, msg = None): # Print list of new objs, making sure that the list is # correctly garbage-collected by the GC print "\n# -- %s:" % (msg or "New objects") self.scan(self._print_new_obj) print "# ---------------\n" # # Method 2: Keep track of the garbage list # Advantages: we have live pointers to the new live objects. And fast # Drawbacks: will only show the object /after/ the GC had tried to # reclaim them, not as soon as they have been # creaded. Still useful to debug leaks... But: are we sure # that lost objects are only found in cycles ??? Same # type restrictions as for method 1 ??? #
def test_it(collectible): dd() dd('======= ', ('collectible' if collectible else 'uncollectible'), ' object =======') dd() gc.collect() dd('*** init, nr of referrers: ', len(gc.get_referrers(One))) dd(' garbage: ', gc.garbage) one = One(collectible) dd(' created: ', one.typ, ': ', one) dd(' nr of referrers: ', len(gc.get_referrers(One))) dd(' delete:') del one gc.collect() dd('*** after gc, nr of referrers: ', len(gc.get_referrers(One))) dd(' garbage: ', gc.garbage)
def package(): # Apply gevent monkey patches as early as possible during tests. from gevent.monkey import patch_all patch_all() # Enable all warnings in test mode. warnings.resetwarnings() warnings.simplefilter('default') # Look for memory leaks. gc.set_debug(gc.DEBUG_UNCOLLECTABLE) yield None # Print memory leaks. if gc.garbage: # pragma: no cover print('Uncollectable objects found:') for obj in gc.garbage: print(obj)
def test_finalizer(self): # A() is uncollectable if it is part of a cycle, make sure it shows up # in gc.garbage. class A: def __del__(self): pass class B: pass a = A() a.a = a id_a = id(a) b = B() b.b = b gc.collect() del a del b self.assertNotEqual(gc.collect(), 0) for obj in gc.garbage: if id(obj) == id_a: del obj.a break else: self.fail("didn't find obj in garbage (finalizer)") gc.garbage.remove(obj)
def test_finalizer_newclass(self): # A() is uncollectable if it is part of a cycle, make sure it shows up # in gc.garbage. class A(object): def __del__(self): pass class B(object): pass a = A() a.a = a id_a = id(a) b = B() b.b = b gc.collect() del a del b self.assertNotEqual(gc.collect(), 0) for obj in gc.garbage: if id(obj) == id_a: del obj.a break else: self.fail("didn't find obj in garbage (finalizer)") gc.garbage.remove(obj)
def test_saveall(self): # Verify that cyclic garbage like lists show up in gc.garbage if the # SAVEALL option is enabled. # First make sure we don't save away other stuff that just happens to # be waiting for collection. gc.collect() # if this fails, someone else created immortal trash self.assertEqual(gc.garbage, []) L = [] L.append(L) id_L = id(L) debug = gc.get_debug() gc.set_debug(debug | gc.DEBUG_SAVEALL) del L gc.collect() gc.set_debug(debug) self.assertEqual(len(gc.garbage), 1) obj = gc.garbage.pop() self.assertEqual(id(obj), id_L)
def test_boom(self): class Boom: def __getattr__(self, someattribute): del self.attr raise AttributeError a = Boom() b = Boom() a.attr = b b.attr = a gc.collect() garbagelen = len(gc.garbage) del a, b # a<->b are in a trash cycle now. Collection will invoke # Boom.__getattr__ (to see whether a and b have __del__ methods), and # __getattr__ deletes the internal "attr" attributes as a side effect. # That causes the trash cycle to get reclaimed via refcounts falling to # 0, thus mutating the trash graph as a side effect of merely asking # whether __del__ exists. This used to (before 2.3b1) crash Python. # Now __getattr__ isn't called. self.assertEqual(gc.collect(), 4) self.assertEqual(len(gc.garbage), garbagelen)
def test_boom2_new(self): class Boom2_New(object): def __init__(self): self.x = 0 def __getattr__(self, someattribute): self.x += 1 if self.x > 1: del self.attr raise AttributeError a = Boom2_New() b = Boom2_New() a.attr = b b.attr = a gc.collect() garbagelen = len(gc.garbage) del a, b self.assertEqual(gc.collect(), 4) self.assertEqual(len(gc.garbage), garbagelen)
def test_main(): enabled = gc.isenabled() gc.disable() assert not gc.isenabled() debug = gc.get_debug() gc.set_debug(debug & ~gc.DEBUG_LEAK) # this test is supposed to leak try: gc.collect() # Delete 2nd generation garbage run_unittest(GCTests, GCTogglingTests) finally: gc.set_debug(debug) # test gc.enable() even if GC is disabled by default if verbose: print("restoring automatic collection") # make sure to always test gc.enable() gc.enable() assert gc.isenabled() if not enabled: gc.disable()
def test_write(self): delta = 0 rows = [[1,2,3]]*5 s = NUL() lastrc = sys.gettotalrefcount() for i in range(20): gc.collect() self.assertEqual(gc.garbage, []) rc = sys.gettotalrefcount() writer = csv.writer(s) for row in rows: writer.writerow(row) delta = rc-lastrc lastrc = rc # if writer leaks during write, last delta should be 5 or more self.assertEqual(delta < 5, True)
def test_boom_new(self): # boom__new and boom2_new are exactly like boom and boom2, except use # new-style classes. class Boom_New(object): def __getattr__(self, someattribute): del self.attr raise AttributeError a = Boom_New() b = Boom_New() a.attr = b b.attr = a gc.collect() garbagelen = len(gc.garbage) del a, b self.assertEqual(gc.collect(), 4) self.assertEqual(len(gc.garbage), garbagelen)
def test_main(): enabled = gc.isenabled() gc.disable() assert not gc.isenabled() debug = gc.get_debug() gc.set_debug(debug & ~gc.DEBUG_LEAK) # this test is supposed to leak try: gc.collect() # Delete 2nd generation garbage run_unittest(GCTests, GCTogglingTests) finally: gc.set_debug(debug) # test gc.enable() even if GC is disabled by default if verbose: print "restoring automatic collection" # make sure to always test gc.enable() gc.enable() assert gc.isenabled() if not enabled: gc.disable()
def tearDown(self): # Restore gc state del self.visit gc.callbacks.remove(self.cb1) gc.callbacks.remove(self.cb2) gc.set_debug(self.debug) if self.enabled: gc.enable() # destroy any uncollectables gc.collect() for obj in gc.garbage: if isinstance(obj, Uncollectable): obj.partner = None del gc.garbage[:] del self.othergarbage gc.collect()
def test_main(): enabled = gc.isenabled() gc.disable() assert not gc.isenabled() debug = gc.get_debug() gc.set_debug(debug & ~gc.DEBUG_LEAK) # this test is supposed to leak try: gc.collect() # Delete 2nd generation garbage run_unittest(GCTests, GCTogglingTests, GCCallbackTests) finally: gc.set_debug(debug) # test gc.enable() even if GC is disabled by default if verbose: print("restoring automatic collection") # make sure to always test gc.enable() gc.enable() assert gc.isenabled() if not enabled: gc.disable()
def __init__(self, interval=1.0, debug=False): """ Initializes garbage collector @param interval float: timeout interval in seconds. Default: 1s @param debug bool: debug output. Default: False """ super().__init__() self.debug = debug if debug: gc.set_debug(gc.DEBUG_LEAK) self.timer = QTimer() self.timer.timeout.connect(self.check) self.threshold = gc.get_threshold() gc.disable() self.timer.start(interval * 1000)
def check(self): """ Method called by the garbage collector timer to check if there is something to collect. """ # return self.debug_cycles() # uncomment to just debug cycles l0, l1, l2 = gc.get_count() if self.debug: logger.debug('gc_check called: {0} {1} {2}'.format(l0, l1, l2)) if l0 > self.threshold[0]: num = gc.collect(0) if self.debug: logger.debug('collecting gen 0, found: {0:d} unreachable' ''.format(num)) if l1 > self.threshold[1]: num = gc.collect(1) if self.debug: logger.debug('collecting gen 1, found: {0:d} unreachable' ''.format(num)) if l2 > self.threshold[2]: num = gc.collect(2) if self.debug: logger.debug('collecting gen 2, found: {0:d} ' 'unreachable'.format(num))
def test_garbageCollectedTransactionAborts(self): """ When an L{IAsyncTransaction} is garbage collected, it ought to abort itself. """ t = self.createTransaction() self.resultOf(t.execSQL("echo", [])) conns = self.factory.connections self.assertEquals(len(conns), 1) self.assertEquals(conns[0]._rollbackCount, 0) del t gc.collect() self.flushHolders() self.assertEquals(len(conns), 1) self.assertEquals(conns[0]._rollbackCount, 1) self.assertEquals(conns[0]._commitCount, 0)
def test_legacy_finalizer(self): # A() is uncollectable if it is part of a cycle, make sure it shows up # in gc.garbage. @with_tp_del class A: def __tp_del__(self): pass class B: pass a = A() a.a = a id_a = id(a) b = B() b.b = b gc.collect() del a del b self.assertNotEqual(gc.collect(), 0) for obj in gc.garbage: if id(obj) == id_a: del obj.a break else: self.fail("didn't find obj in garbage (finalizer)") gc.garbage.remove(obj)
def test_legacy_finalizer_newclass(self): # A() is uncollectable if it is part of a cycle, make sure it shows up # in gc.garbage. @with_tp_del class A(object): def __tp_del__(self): pass class B(object): pass a = A() a.a = a id_a = id(a) b = B() b.b = b gc.collect() del a del b self.assertNotEqual(gc.collect(), 0) for obj in gc.garbage: if id(obj) == id_a: del obj.a break else: self.fail("didn't find obj in garbage (finalizer)") gc.garbage.remove(obj)
