[Jython-checkins] jython: Adjusted test_gc.py and test_gc_jy.py such that they also pass in CPython (i.e.
stefan.richthofer
jython-checkins at python.org
Tue Feb 17 03:59:45 CET 2015
https://hg.python.org/jython/rev/efda62bbec80
changeset: 7585:efda62bbec80
user: Stefan Richthofer <stefan.richthofer at gmx.de>
date: Tue Feb 17 03:59:27 2015 +0100
summary:
Adjusted test_gc.py and test_gc_jy.py such that they also pass in CPython (i.e. isolated Jython-specific code by skips or try-blocks)
files:
Lib/test/test_gc.py | 259 +++++++++++++++-------------
Lib/test/test_gc_jy.py | 209 +++++++++++-----------
2 files changed, 236 insertions(+), 232 deletions(-)
diff --git a/Lib/test/test_gc.py b/Lib/test/test_gc.py
--- a/Lib/test/test_gc.py
+++ b/Lib/test/test_gc.py
@@ -1,9 +1,10 @@
import unittest
-#from test.test_support import verbose, run_unittest
+from test.test_support import verbose, run_unittest
from test import test_support
import sys
import gc
import weakref
+import time
try:
import threading
@@ -59,6 +60,7 @@
@classmethod
def tearDownClass(cls):
+ #Jython-specific block:
try:
gc.setJythonGCFlags(cls.savedJythonGCFlags)
gc.stopMonitoring()
@@ -203,17 +205,12 @@
self.assertEqual(gc.collect(), 2)
def test_frame(self):
- flg = gc.getJythonGCFlags()
- #gc.addJythonGCFlags(gc.VERBOSE)
- #sporadically fails in Jython, no idea why.
def f():
frame = sys._getframe()
gc.collect()
f()
col = gc.collect()
- gc.setJythonGCFlags(flg)
self.assertEqual(col, 1)
-
def test_saveall(self):
# Verify that cyclic garbage like lists show up in gc.garbage if the
@@ -316,104 +313,115 @@
gc.collect(2)
assertEqual(gc.get_count(), (0, 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(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()}
+ try:
+ gc.disable()
+ except NotImplementedError:
+ #i.e. Jython is running (or other non-CPython interpreter without gc-disabling)
+ pass
-# @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_checkinterval = sys.getcheckinterval()
-# sys.setcheckinterval(3)
-# try:
-# exit = False
-# threads = []
-# for i in range(N_THREADS):
-# t = threading.Thread(target=run_thread)
-# threads.append(t)
-# for t in threads:
-# t.start()
-# time.sleep(1.0)
-# exit = True
-# for t in threads:
-# t.join()
-# finally:
-# pass
-# sys.setcheckinterval(old_checkinterval)
-# gc.collect()
-# self.assertEqual(len(C.inits), len(C.dels))
+ @unittest.skipIf(test_support.is_jython,
+ '''
+ Jython does not have a trashcan mechanism.
+ This test should still not fail but currently does.
+ This is because the massive referencing in this test brings
+ sync gc emulation to its limit. Making this more robust is
+ of no priority for now.
+ ''')
+ @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_checkinterval = sys.getcheckinterval()
+ sys.setcheckinterval(3)
+ try:
+ exit = False
+ threads = []
+ for i in range(N_THREADS):
+ t = threading.Thread(target=run_thread)
+ threads.append(t)
+ for t in threads:
+ t.start()
+ time.sleep(0.01)
+ exit = True
+ for t in threads:
+ t.join()
+ finally:
+ pass
+ sys.setcheckinterval(old_checkinterval)
+ self.assertEqual(len(C.inits), len(C.dels))
def test_boom(self):
class Boom:
@@ -733,28 +741,29 @@
# empty __dict__.
self.assertEqual(x, None)
-# def test_main():
-# unittest.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_main():
+ enabled = gc.isenabled()
+ try:
+ gc.disable()
+ assert not gc.isenabled()
+ except NotImplementedError:
+ pass
+ 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()
if __name__ == "__main__":
- #test_main()
unittest.main()
diff --git a/Lib/test/test_gc_jy.py b/Lib/test/test_gc_jy.py
--- a/Lib/test/test_gc_jy.py
+++ b/Lib/test/test_gc_jy.py
@@ -1,31 +1,21 @@
"""
Tests some Jython-specific gc aspects and debugging
features.
+Skips and try-blocks assure that this test-script is
+still runnable with CPython and passes there as far
+as not skipped.
"""
import unittest
-#from test.test_support import verbose, run_unittest
-#import sys
+from test import test_support
import time
import gc
import weakref
-from java.lang import System, Runnable
-
-# class FinalizationDummy:
-# def __del__(self):
-# time.sleep(3.5)
-# print "FinalizationDummy.__del__"
-# time.sleep(3.5)
-#
-# class ResurrectionDummy:
-# def __del__(self):
-# print "ResurrectionDummy.__del__"
-# ResurrectionDummy.resurrected = self.toResurrect
-#
-# class SelfResurrectionDummy:
-# def __del__(self):
-# print "SelfResurrectionDummy.__del__"
-# SelfResurrectionDummy.resurrected = self
+try:
+ from java.lang import System, Runnable
+except ImportError:
+ #i.e. Jython is running
+ pass
class GCTests_Jy_CyclicGarbage(unittest.TestCase):
@@ -101,6 +91,8 @@
self.fail("didn't find obj in garbage (finalizer)")
gc.garbage.remove(obj)
+ @unittest.skipUnless(test_support.is_jython,
+ 'CPython has no monitor state')
def test_manual_monitoring(self):
# since tuples are immutable we close the loop with a list
l = []
@@ -117,6 +109,8 @@
self.assertEqual(gc.collect(), 1)
+ at unittest.skipUnless(test_support.is_jython,
+ 'CPython has no gc preprocess and postprocess features')
class GCTests_Jy_preprocess_and_postprocess(unittest.TestCase):
def test_finalization_preprocess_and_postprocess(self):
@@ -156,6 +150,82 @@
prePr = PreProcess()
postPr = PostProcess()
time.sleep(1) # <- to avoid that the newly registered processes
+ # become subject to previous run (remember: We
+ # are not in monitor-mode, i.e. gc runs async.
+ gc.registerPreFinalizationProcess(prePr)
+ gc.registerPostFinalizationProcess(postPr)
+ #Note that order matters here:
+ #If the flag gc.DONT_FINALIZE_RESURRECTED_OBJECTS is used,
+ #gc.registerPostFinalizationProcess(postPr, 0) would lead to failure,
+ #because postPr asserts that a's finalizer already ran. Since
+ #DONT_FINALIZE_RESURRECTED_OBJECTS also inserted a postprocess,
+ #to perform delayed finalization, the 0-index would prepend postPr
+ #before the process that actually runs the finalizers.
+ System.gc()
+ #we wait a bit longer here, since PostProcess runs asynchronous
+ #and must wait for the finalizer of A
+ time.sleep(2)
+ self.assertIn("run PostProcess", comments)
+ comments = []
+ gc.unregisterPreFinalizationProcess(prePr)
+ gc.unregisterPostFinalizationProcess(postPr)
+
+
+ at unittest.skipUnless(test_support.is_jython,
+ 'This class tests detailed Jython-specific behavior.')
+class GCTests_Jy_Delayed_Finalization(unittest.TestCase):
+
+ @classmethod
+ def setUpClass(cls):
+ #Jython-specific block:
+ try:
+ cls.savedJythonGCFlags = gc.getJythonGCFlags()
+ #the finalizer-related tests need this flag to pass in Jython:
+ gc.addJythonGCFlags(gc.DONT_FINALIZE_RESURRECTED_OBJECTS)
+ gc.stopMonitoring()
+ except Exception:
+ pass
+
+ @classmethod
+ def tearDownClass(cls):
+ try:
+ gc.setJythonGCFlags(cls.savedJythonGCFlags)
+ except Exception:
+ pass
+
+ def test_finalization_preprocess_and_postprocess(self):
+ #Note that this test is done here again (already was in another class
+ #in this module), to see that everything works as it should also with
+ #a different flag-context.
+ comments = []
+ self0 = self
+ class A:
+ def __del__(self):
+ self0.assertIn("run PreProcess", comments)
+ comments.append("A del")
+ #let's simulate a time-consuming finalizer
+ #to ensure that post finalization processing
+ #is sensitive to this
+ time.sleep(0.5)
+ comments.append("A del done")
+
+ class PreProcess(Runnable):
+ def run(self):
+ self0.assertEqual(comments, [])
+ comments.append("run PreProcess")
+
+ class PostProcess(Runnable):
+ def run(self):
+ self0.assertIn("run PreProcess", comments)
+ self0.assertIn("A del", comments)
+ self0.assertIn("A del done", comments)
+ comments.append("run PostProcess")
+
+ a = A()
+ a = None
+ prePr = PreProcess()
+ postPr = PostProcess()
+ time.sleep(1) # <- to avoid that the newly registered processes
# become subject to previous run
gc.registerPreFinalizationProcess(prePr)
gc.registerPostFinalizationProcess(postPr)
@@ -176,86 +246,7 @@
gc.unregisterPostFinalizationProcess(postPr)
-class GCTests_Jy_Delayed_Finalization(unittest.TestCase):
-
- @classmethod
- def setUpClass(cls):
- #Jython-specific block:
- try:
- cls.savedJythonGCFlags = gc.getJythonGCFlags()
- #the finalizer-related tests need this flag to pass in Jython:
- gc.addJythonGCFlags(gc.DONT_FINALIZE_RESURRECTED_OBJECTS)
- gc.stopMonitoring()
- except Exception:
- pass
-
- @classmethod
- def tearDownClass(cls):
- try:
- gc.setJythonGCFlags(cls.savedJythonGCFlags)
- except Exception:
- pass
-
- def test_finalization_preprocess_and_postprocess(self):
- #print "test_finalization_preprocess_and_postprocess"
- #Note that this test is done here again (already was in another class
- #in this module), to see that everything works as it should also with
- #a different flag-context.
- #print "test_finalization_preprocess_and_postprocess"
- #gc.removeJythonGCFlags(gc.DONT_FINALIZE_RESURRECTED_OBJECTS)
- comments = []
- self0 = self
- class A:
- def __del__(self):
- #print "del A"
- self0.assertIn("run PreProcess", comments)
- comments.append("A del")
- #let's simulate a time-consuming finalizer
- #to ensure that post finalization processing
- #is sensitive to this
- time.sleep(0.5)
- comments.append("A del done")
-
- class PreProcess(Runnable):
- def run(self):
- self0.assertEqual(comments, [])
- comments.append("run PreProcess")
-
- class PostProcess(Runnable):
- def run(self):
- self0.assertIn("run PreProcess", comments)
- self0.assertIn("A del", comments)
- self0.assertIn("A del done", comments)
- comments.append("run PostProcess")
-
- a = A()
- a = None
- prePr = PreProcess()
- postPr = PostProcess()
- time.sleep(1) # <- to avoid that the newly registered processes
- # become subject to previous run
- gc.registerPreFinalizationProcess(prePr)
- gc.registerPostFinalizationProcess(postPr)
- #Note that order matters here:
- #If the flag gc.DONT_FINALIZE_RESURRECTED_OBJECTS is used,
- #gc.registerPostFinalizationProcess(postPr, 0) would lead to failure,
- #because postPr asserts that a's finalizer already ran. Since
- #DONT_FINALIZE_RESURRECTED_OBJECTS also inserted a postprocess,
- #to perform delayed finalization, the 0-index would prepend postPr
- #before the process that actually runs the finalizers.
- System.gc()
- #we wait a bit longer here, since PostProcess runs asynchronous
- #and must wait for the finalizer of A
- time.sleep(2)
- self.assertIn("run PostProcess", comments)
- comments = []
- gc.unregisterPreFinalizationProcess(prePr)
- gc.unregisterPostFinalizationProcess(postPr)
-
-
def test_delayedFinalization(self):
- #gc.addJythonGCFlags(gc.DONT_FINALIZE_RESURRECTED_OBJECTS)
- #gc.addJythonGCFlags(gc.VERBOSE)
resurrect = []
comments = []
@@ -291,15 +282,17 @@
del c
self.assertNotEqual(gc.collect(), 0)
time.sleep(1)
- #print comments
- #print resurrect
+ #Note that CPython would collect a, b and c in one run.
+ #With gc.DONT_FINALIZE_RESURRECTED_OBJECTS set, Jython
+ #Would not collect a and b in the same run with c
+ #because a and b might have been resurrected by c and
+ #Java allows not to detect such resurrection in any
+ #other way than waiting for the next gc-run.
self.assertIn('del c', comments)
self.assertEqual(1, len(comments))
comments = []
self.assertNotEqual(gc.collect(), 0)
time.sleep(1)
- #print comments
- #print resurrect
self.assertIn('del a', comments)
self.assertEqual(1, len(comments))
comments = []
@@ -307,8 +300,6 @@
time.sleep(1)
self.assertIn('del b', comments)
self.assertEqual(1, len(comments))
- #gc.removeJythonGCFlags(gc.DONT_FINALIZE_RESURRECTED_OBJECTS)
- #gc.removeJythonGCFlags(gc.VERBOSE)
class GCTests_Jy_Monitoring(unittest.TestCase):
@@ -336,6 +327,7 @@
except Exception:
pass
+ @unittest.skipUnless(test_support.is_jython, 'CPython has no monitor-state.')
def test_monitor_status_after_delayed_finalization(self):
resurrect = []
comments = []
@@ -409,8 +401,11 @@
a.b.lst = lst
del lst
del lst1
- self.assertTrue(gc.isMonitored(a))
- self.assertTrue(gc.isMonitored(a.b))
+ try:
+ self.assertTrue(gc.isMonitored(a))
+ self.assertTrue(gc.isMonitored(a.b))
+ except AttributeError:
+ pass
del a
self.assertEqual(gc.collect(), 2) # c is not cyclic and a, b are resurrected,
# the cycle of two lists is counted here
@@ -543,6 +538,7 @@
except Exception:
pass
+ @unittest.skipUnless(test_support.is_jython, '')
def test_weakref_after_resurrection_threadsafe(self):
resurrect = []
comments = []
@@ -587,7 +583,6 @@
# restored or the deletion is confirmed.
except Exception:
pass
- #self.assertEqual(str(wa()), '<a>')
self.assertEqual(comments, [])
self.assertEqual(resurrect, [])
while comments == [] or resurrect == []:
--
Repository URL: https://hg.python.org/jython
More information about the Jython-checkins
mailing list