mirror of
https://github.com/jart/cosmopolitan.git
synced 2025-01-31 19:43:32 +00:00
5ef64dbcdb
These are the commits from https://github.com/ahgamut/cpython/tree/cosmo_py36 squashed for simplicity. Also included is the pyconfig.h used for compilation. The pyconfig.h has to be changed manually in case Cosmopolitan gets new features.
1047 lines
34 KiB
Python
1047 lines
34 KiB
Python
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
|
|
|
|
try:
|
|
import _thread
|
|
import threading
|
|
except ImportError:
|
|
threading = None
|
|
|
|
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()
|
|
|
|
@unittest.skipUnless(threading, "test meaningless on builds without threads")
|
|
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 "<X %%r>" %% 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"<X 'first'>", 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"[<X 'first'>, <X 'second'>]" in stderr) or
|
|
(b"[<X 'second'>, <X 'first'>]" 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)
|
|
|
|
|
|
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()
|