import unittest from test.support import (verbose, refcount_test, run_unittest, strip_python_stderr, cpython_only, start_threads, temp_dir, requires_type_collecting, TESTFN, unlink) from test.support.script_helper import assert_python_ok, make_script import sys import time import gc import weakref import threading try: from _testcapi import with_tp_del except ImportError: def with_tp_del(cls): class C(object): def __new__(cls, *args, **kwargs): raise TypeError('requires _testcapi.with_tp_del') return C ### Support code ############################################################################### # Bug 1055820 has several tests of longstanding bugs involving weakrefs and # cyclic gc. # An instance of C1055820 has a self-loop, so becomes cyclic trash when # unreachable. class C1055820(object): def __init__(self, i): self.i = i self.loop = self class GC_Detector(object): # Create an instance I. Then gc hasn't happened again so long as # I.gc_happened is false. def __init__(self): self.gc_happened = False def it_happened(ignored): self.gc_happened = True # Create a piece of cyclic trash that triggers it_happened when # gc collects it. self.wr = weakref.ref(C1055820(666), it_happened) @with_tp_del class Uncollectable(object): """Create a reference cycle with multiple __del__ methods. An object in a reference cycle will never have zero references, and so must be garbage collected. If one or more objects in the cycle have __del__ methods, the gc refuses to guess an order, and leaves the cycle uncollected.""" def __init__(self, partner=None): if partner is None: self.partner = Uncollectable(partner=self) else: self.partner = partner def __tp_del__(self): pass ### Tests ############################################################################### class GCTests(unittest.TestCase): def test_list(self): l = [] l.append(l) gc.collect() del l self.assertEqual(gc.collect(), 1) def test_dict(self): d = {} d[1] = d gc.collect() del d self.assertEqual(gc.collect(), 1) def test_tuple(self): # since tuples are immutable we close the loop with a list l = [] t = (l,) l.append(t) gc.collect() del t del l self.assertEqual(gc.collect(), 2) def test_class(self): class A: pass A.a = A gc.collect() del A self.assertNotEqual(gc.collect(), 0) def test_newstyleclass(self): class A(object): pass gc.collect() del A self.assertNotEqual(gc.collect(), 0) def test_instance(self): class A: pass a = A() a.a = a gc.collect() del a self.assertNotEqual(gc.collect(), 0) @requires_type_collecting def test_newinstance(self): class A(object): pass a = A() a.a = a gc.collect() del a self.assertNotEqual(gc.collect(), 0) class B(list): pass class C(B, A): pass a = C() a.a = a gc.collect() del a self.assertNotEqual(gc.collect(), 0) del B, C self.assertNotEqual(gc.collect(), 0) A.a = A() del A self.assertNotEqual(gc.collect(), 0) self.assertEqual(gc.collect(), 0) def test_method(self): # Tricky: self.__init__ is a bound method, it references the instance. class A: def __init__(self): self.init = self.__init__ a = A() gc.collect() del a self.assertNotEqual(gc.collect(), 0) @cpython_only 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) @cpython_only 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) def test_function(self): # Tricky: f -> d -> f, code should call d.clear() after the exec to # break the cycle. d = {} exec("def f(): pass\n", d) gc.collect() del d self.assertEqual(gc.collect(), 2) @refcount_test def test_frame(self): def f(): frame = sys._getframe() gc.collect() f() self.assertEqual(gc.collect(), 1) 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_del(self): # __del__ methods can trigger collection, make this to happen thresholds = gc.get_threshold() gc.enable() gc.set_threshold(1) class A: def __del__(self): dir(self) a = A() del a gc.disable() gc.set_threshold(*thresholds) def test_del_newclass(self): # __del__ methods can trigger collection, make this to happen thresholds = gc.get_threshold() gc.enable() gc.set_threshold(1) class A(object): def __del__(self): dir(self) a = A() del a gc.disable() gc.set_threshold(*thresholds) # The following two tests are fragile: # They precisely count the number of allocations, # which is highly implementation-dependent. # For example, disposed tuples are not freed, but reused. # To minimize variations, though, we first store the get_count() results # and check them at the end. @refcount_test def test_get_count(self): gc.collect() a, b, c = gc.get_count() x = [] d, e, f = gc.get_count() self.assertEqual((b, c), (0, 0)) self.assertEqual((e, f), (0, 0)) # This is less fragile than asserting that a equals 0. self.assertLess(a, 5) # Between the two calls to get_count(), at least one object was # created (the list). self.assertGreater(d, a) @refcount_test def test_collect_generations(self): gc.collect() # This object will "trickle" into generation N + 1 after # each call to collect(N) x = [] gc.collect(0) # x is now in gen 1 a, b, c = gc.get_count() gc.collect(1) # x is now in gen 2 d, e, f = gc.get_count() gc.collect(2) # x is now in gen 3 g, h, i = gc.get_count() # We don't check a, d, g since their exact values depends on # internal implementation details of the interpreter. self.assertEqual((b, c), (1, 0)) self.assertEqual((e, f), (0, 1)) self.assertEqual((h, i), (0, 0)) def test_trashcan(self): class Ouch: n = 0 def __del__(self): Ouch.n = Ouch.n + 1 if Ouch.n % 17 == 0: gc.collect() # "trashcan" is a hack to prevent stack overflow when deallocating # very deeply nested tuples etc. It works in part by abusing the # type pointer and refcount fields, and that can yield horrible # problems when gc tries to traverse the structures. # If this test fails (as it does in 2.0, 2.1 and 2.2), it will # most likely die via segfault. # Note: In 2.3 the possibility for compiling without cyclic gc was # removed, and that in turn allows the trashcan mechanism to work # via much simpler means (e.g., it never abuses the type pointer or # refcount fields anymore). Since it's much less likely to cause a # problem now, the various constants in this expensive (we force a lot # of full collections) test are cut back from the 2.2 version. gc.enable() N = 150 for count in range(2): t = [] for i in range(N): t = [t, Ouch()] u = [] for i in range(N): u = [u, Ouch()] v = {} for i in range(N): v = {1: v, 2: Ouch()} gc.disable() def test_trashcan_threads(self): # Issue #13992: trashcan mechanism should be thread-safe NESTING = 60 N_THREADS = 2 def sleeper_gen(): """A generator that releases the GIL when closed or dealloc'ed.""" try: yield finally: time.sleep(0.000001) class C(list): # Appending to a list is atomic, which avoids the use of a lock. inits = [] dels = [] def __init__(self, alist): self[:] = alist C.inits.append(None) def __del__(self): # This __del__ is called by subtype_dealloc(). C.dels.append(None) # `g` will release the GIL when garbage-collected. This # helps assert subtype_dealloc's behaviour when threads # switch in the middle of it. g = sleeper_gen() next(g) # Now that __del__ is finished, subtype_dealloc will proceed # to call list_dealloc, which also uses the trashcan mechanism. def make_nested(): """Create a sufficiently nested container object so that the trashcan mechanism is invoked when deallocating it.""" x = C([]) for i in range(NESTING): x = [C([x])] del x def run_thread(): """Exercise make_nested() in a loop.""" while not exit: make_nested() old_switchinterval = sys.getswitchinterval() sys.setswitchinterval(1e-5) try: exit = [] threads = [] for i in range(N_THREADS): t = threading.Thread(target=run_thread) threads.append(t) with start_threads(threads, lambda: exit.append(1)): time.sleep(1.0) finally: sys.setswitchinterval(old_switchinterval) gc.collect() self.assertEqual(len(C.inits), len(C.dels)) 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(self): class Boom2: def __init__(self): self.x = 0 def __getattr__(self, someattribute): self.x += 1 if self.x > 1: del self.attr raise AttributeError a = Boom2() b = Boom2() a.attr = b b.attr = a gc.collect() garbagelen = len(gc.garbage) del a, b # Much like test_boom(), except that __getattr__ doesn't break the # cycle until the second time gc checks for __del__. As of 2.3b1, # there isn't a second time, so this simply cleans up the trash cycle. # We expect a, b, a.__dict__ and b.__dict__ (4 objects) to get # reclaimed this way. self.assertEqual(gc.collect(), 4) self.assertEqual(len(gc.garbage), garbagelen) 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_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_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_is_tracked(self): # Atomic built-in types are not tracked, user-defined objects and # mutable containers are. # NOTE: types with special optimizations (e.g. tuple) have tests # in their own test files instead. self.assertFalse(gc.is_tracked(None)) self.assertFalse(gc.is_tracked(1)) self.assertFalse(gc.is_tracked(1.0)) self.assertFalse(gc.is_tracked(1.0 + 5.0j)) self.assertFalse(gc.is_tracked(True)) self.assertFalse(gc.is_tracked(False)) self.assertFalse(gc.is_tracked(b"a")) self.assertFalse(gc.is_tracked("a")) self.assertFalse(gc.is_tracked(bytearray(b"a"))) self.assertFalse(gc.is_tracked(type)) self.assertFalse(gc.is_tracked(int)) self.assertFalse(gc.is_tracked(object)) self.assertFalse(gc.is_tracked(object())) class UserClass: pass class UserInt(int): pass # Base class is object; no extra fields. class UserClassSlots: __slots__ = () # Base class is fixed size larger than object; no extra fields. class UserFloatSlots(float): __slots__ = () # Base class is variable size; no extra fields. class UserIntSlots(int): __slots__ = () self.assertTrue(gc.is_tracked(gc)) self.assertTrue(gc.is_tracked(UserClass)) self.assertTrue(gc.is_tracked(UserClass())) self.assertTrue(gc.is_tracked(UserInt())) self.assertTrue(gc.is_tracked([])) self.assertTrue(gc.is_tracked(set())) self.assertFalse(gc.is_tracked(UserClassSlots())) self.assertFalse(gc.is_tracked(UserFloatSlots())) self.assertFalse(gc.is_tracked(UserIntSlots())) def test_bug1055820b(self): # Corresponds to temp2b.py in the bug report. ouch = [] def callback(ignored): ouch[:] = [wr() for wr in WRs] Cs = [C1055820(i) for i in range(2)] WRs = [weakref.ref(c, callback) for c in Cs] c = None gc.collect() self.assertEqual(len(ouch), 0) # Make the two instances trash, and collect again. The bug was that # the callback materialized a strong reference to an instance, but gc # cleared the instance's dict anyway. Cs = None gc.collect() self.assertEqual(len(ouch), 2) # else the callbacks didn't run for x in ouch: # If the callback resurrected one of these guys, the instance # would be damaged, with an empty __dict__. self.assertEqual(x, None) def test_bug21435(self): # This is a poor test - its only virtue is that it happened to # segfault on Tim's Windows box before the patch for 21435 was # applied. That's a nasty bug relying on specific pieces of cyclic # trash appearing in exactly the right order in finalize_garbage()'s # input list. # But there's no reliable way to force that order from Python code, # so over time chances are good this test won't really be testing much # of anything anymore. Still, if it blows up, there's _some_ # problem ;-) gc.collect() class A: pass class B: def __init__(self, x): self.x = x def __del__(self): self.attr = None def do_work(): a = A() b = B(A()) a.attr = b b.attr = a do_work() gc.collect() # this blows up (bad C pointer) when it fails @cpython_only def test_garbage_at_shutdown(self): import subprocess code = """if 1: import gc import _testcapi @_testcapi.with_tp_del class X: def __init__(self, name): self.name = name def __repr__(self): return "" %% self.name def __tp_del__(self): pass x = X('first') x.x = x x.y = X('second') del x gc.set_debug(%s) """ def run_command(code): p = subprocess.Popen([sys.executable, "-Wd", "-c", code], stdout=subprocess.PIPE, stderr=subprocess.PIPE) stdout, stderr = p.communicate() p.stdout.close() p.stderr.close() self.assertEqual(p.returncode, 0) self.assertEqual(stdout.strip(), b"") return strip_python_stderr(stderr) stderr = run_command(code % "0") self.assertIn(b"ResourceWarning: gc: 2 uncollectable objects at " b"shutdown; use", stderr) self.assertNotIn(b"", stderr) # With DEBUG_UNCOLLECTABLE, the garbage list gets printed stderr = run_command(code % "gc.DEBUG_UNCOLLECTABLE") self.assertIn(b"ResourceWarning: gc: 2 uncollectable objects at " b"shutdown", stderr) self.assertTrue( (b"[, ]" in stderr) or (b"[, ]" in stderr), stderr) # With DEBUG_SAVEALL, no additional message should get printed # (because gc.garbage also contains normally reclaimable cyclic # references, and its elements get printed at runtime anyway). stderr = run_command(code % "gc.DEBUG_SAVEALL") self.assertNotIn(b"uncollectable objects at shutdown", stderr) @requires_type_collecting def test_gc_main_module_at_shutdown(self): # Create a reference cycle through the __main__ module and check # it gets collected at interpreter shutdown. code = """if 1: class C: def __del__(self): print('__del__ called') l = [C()] l.append(l) """ rc, out, err = assert_python_ok('-c', code) self.assertEqual(out.strip(), b'__del__ called') @requires_type_collecting def test_gc_ordinary_module_at_shutdown(self): # Same as above, but with a non-__main__ module. with temp_dir() as script_dir: module = """if 1: class C: def __del__(self): print('__del__ called') l = [C()] l.append(l) """ code = """if 1: import sys sys.path.insert(0, %r) import gctest """ % (script_dir,) make_script(script_dir, 'gctest', module) rc, out, err = assert_python_ok('-c', code) self.assertEqual(out.strip(), b'__del__ called') @requires_type_collecting def test_global_del_SystemExit(self): code = """if 1: class ClassWithDel: def __del__(self): print('__del__ called') a = ClassWithDel() a.link = a raise SystemExit(0)""" self.addCleanup(unlink, TESTFN) with open(TESTFN, 'w') as script: script.write(code) rc, out, err = assert_python_ok(TESTFN) self.assertEqual(out.strip(), b'__del__ called') def test_get_stats(self): stats = gc.get_stats() self.assertEqual(len(stats), 3) for st in stats: self.assertIsInstance(st, dict) self.assertEqual(set(st), {"collected", "collections", "uncollectable"}) self.assertGreaterEqual(st["collected"], 0) self.assertGreaterEqual(st["collections"], 0) self.assertGreaterEqual(st["uncollectable"], 0) # Check that collection counts are incremented correctly if gc.isenabled(): self.addCleanup(gc.enable) gc.disable() old = gc.get_stats() gc.collect(0) new = gc.get_stats() self.assertEqual(new[0]["collections"], old[0]["collections"] + 1) self.assertEqual(new[1]["collections"], old[1]["collections"]) self.assertEqual(new[2]["collections"], old[2]["collections"]) gc.collect(2) new = gc.get_stats() self.assertEqual(new[0]["collections"], old[0]["collections"] + 1) self.assertEqual(new[1]["collections"], old[1]["collections"]) self.assertEqual(new[2]["collections"], old[2]["collections"] + 1) def test_freeze(self): gc.freeze() self.assertGreater(gc.get_freeze_count(), 0) gc.unfreeze() self.assertEqual(gc.get_freeze_count(), 0) class GCCallbackTests(unittest.TestCase): def setUp(self): # Save gc state and disable it. self.enabled = gc.isenabled() gc.disable() self.debug = gc.get_debug() gc.set_debug(0) gc.callbacks.append(self.cb1) gc.callbacks.append(self.cb2) self.othergarbage = [] 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 preclean(self): # Remove all fluff from the system. Invoke this function # manually rather than through self.setUp() for maximum # safety. self.visit = [] gc.collect() garbage, gc.garbage[:] = gc.garbage[:], [] self.othergarbage.append(garbage) self.visit = [] def cb1(self, phase, info): self.visit.append((1, phase, dict(info))) def cb2(self, phase, info): self.visit.append((2, phase, dict(info))) if phase == "stop" and hasattr(self, "cleanup"): # Clean Uncollectable from garbage uc = [e for e in gc.garbage if isinstance(e, Uncollectable)] gc.garbage[:] = [e for e in gc.garbage if not isinstance(e, Uncollectable)] for e in uc: e.partner = None def test_collect(self): self.preclean() gc.collect() # Algorithmically verify the contents of self.visit # because it is long and tortuous. # Count the number of visits to each callback n = [v[0] for v in self.visit] n1 = [i for i in n if i == 1] n2 = [i for i in n if i == 2] self.assertEqual(n1, [1]*2) self.assertEqual(n2, [2]*2) # Count that we got the right number of start and stop callbacks. n = [v[1] for v in self.visit] n1 = [i for i in n if i == "start"] n2 = [i for i in n if i == "stop"] self.assertEqual(n1, ["start"]*2) self.assertEqual(n2, ["stop"]*2) # Check that we got the right info dict for all callbacks for v in self.visit: info = v[2] self.assertTrue("generation" in info) self.assertTrue("collected" in info) self.assertTrue("uncollectable" in info) def test_collect_generation(self): self.preclean() gc.collect(2) for v in self.visit: info = v[2] self.assertEqual(info["generation"], 2) @cpython_only def test_collect_garbage(self): self.preclean() # Each of these cause four objects to be garbage: Two # Uncolectables and their instance dicts. Uncollectable() Uncollectable() C1055820(666) gc.collect() for v in self.visit: if v[1] != "stop": continue info = v[2] self.assertEqual(info["collected"], 2) self.assertEqual(info["uncollectable"], 8) # We should now have the Uncollectables in gc.garbage self.assertEqual(len(gc.garbage), 4) for e in gc.garbage: self.assertIsInstance(e, Uncollectable) # Now, let our callback handle the Uncollectable instances self.cleanup=True self.visit = [] gc.garbage[:] = [] gc.collect() for v in self.visit: if v[1] != "stop": continue info = v[2] self.assertEqual(info["collected"], 0) self.assertEqual(info["uncollectable"], 4) # Uncollectables should be gone self.assertEqual(len(gc.garbage), 0) class GCTogglingTests(unittest.TestCase): def setUp(self): gc.enable() def tearDown(self): gc.disable() def test_bug1055820c(self): # Corresponds to temp2c.py in the bug report. This is pretty # elaborate. c0 = C1055820(0) # Move c0 into generation 2. gc.collect() c1 = C1055820(1) c1.keep_c0_alive = c0 del c0.loop # now only c1 keeps c0 alive c2 = C1055820(2) c2wr = weakref.ref(c2) # no callback! ouch = [] def callback(ignored): ouch[:] = [c2wr()] # The callback gets associated with a wr on an object in generation 2. c0wr = weakref.ref(c0, callback) c0 = c1 = c2 = None # What we've set up: c0, c1, and c2 are all trash now. c0 is in # generation 2. The only thing keeping it alive is that c1 points to # it. c1 and c2 are in generation 0, and are in self-loops. There's a # global weakref to c2 (c2wr), but that weakref has no callback. # There's also a global weakref to c0 (c0wr), and that does have a # callback, and that callback references c2 via c2wr(). # # c0 has a wr with callback, which references c2wr # ^ # | # | Generation 2 above dots #. . . . . . . .|. . . . . . . . . . . . . . . . . . . . . . . . # | Generation 0 below dots # | # | # ^->c1 ^->c2 has a wr but no callback # | | | | # <--v <--v # # So this is the nightmare: when generation 0 gets collected, we see # that c2 has a callback-free weakref, and c1 doesn't even have a # weakref. Collecting generation 0 doesn't see c0 at all, and c0 is # the only object that has a weakref with a callback. gc clears c1 # and c2. Clearing c1 has the side effect of dropping the refcount on # c0 to 0, so c0 goes away (despite that it's in an older generation) # and c0's wr callback triggers. That in turn materializes a reference # to c2 via c2wr(), but c2 gets cleared anyway by gc. # We want to let gc happen "naturally", to preserve the distinction # between generations. junk = [] i = 0 detector = GC_Detector() while not detector.gc_happened: i += 1 if i > 10000: self.fail("gc didn't happen after 10000 iterations") self.assertEqual(len(ouch), 0) junk.append([]) # this will eventually trigger gc self.assertEqual(len(ouch), 1) # else the callback wasn't invoked for x in ouch: # If the callback resurrected c2, the instance would be damaged, # with an empty __dict__. self.assertEqual(x, None) def test_bug1055820d(self): # Corresponds to temp2d.py in the bug report. This is very much like # test_bug1055820c, but uses a __del__ method instead of a weakref # callback to sneak in a resurrection of cyclic trash. ouch = [] class D(C1055820): def __del__(self): ouch[:] = [c2wr()] d0 = D(0) # Move all the above into generation 2. gc.collect() c1 = C1055820(1) c1.keep_d0_alive = d0 del d0.loop # now only c1 keeps d0 alive c2 = C1055820(2) c2wr = weakref.ref(c2) # no callback! d0 = c1 = c2 = None # What we've set up: d0, c1, and c2 are all trash now. d0 is in # generation 2. The only thing keeping it alive is that c1 points to # it. c1 and c2 are in generation 0, and are in self-loops. There's # a global weakref to c2 (c2wr), but that weakref has no callback. # There are no other weakrefs. # # d0 has a __del__ method that references c2wr # ^ # | # | Generation 2 above dots #. . . . . . . .|. . . . . . . . . . . . . . . . . . . . . . . . # | Generation 0 below dots # | # | # ^->c1 ^->c2 has a wr but no callback # | | | | # <--v <--v # # So this is the nightmare: when generation 0 gets collected, we see # that c2 has a callback-free weakref, and c1 doesn't even have a # weakref. Collecting generation 0 doesn't see d0 at all. gc clears # c1 and c2. Clearing c1 has the side effect of dropping the refcount # on d0 to 0, so d0 goes away (despite that it's in an older # generation) and d0's __del__ triggers. That in turn materializes # a reference to c2 via c2wr(), but c2 gets cleared anyway by gc. # We want to let gc happen "naturally", to preserve the distinction # between generations. detector = GC_Detector() junk = [] i = 0 while not detector.gc_happened: i += 1 if i > 10000: self.fail("gc didn't happen after 10000 iterations") self.assertEqual(len(ouch), 0) junk.append([]) # this will eventually trigger gc self.assertEqual(len(ouch), 1) # else __del__ wasn't invoked for x in ouch: # If __del__ resurrected c2, the instance would be damaged, with an # empty __dict__. self.assertEqual(x, None) 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() if __name__ == "__main__": test_main()