ALLOWED_TYPES = ('processes',)
+ def setUp(self):
+ self.registry_backup = util._finalizer_registry.copy()
+ util._finalizer_registry.clear()
+
+ def tearDown(self):
+ self.assertFalse(util._finalizer_registry)
+ util._finalizer_registry.update(self.registry_backup)
+
@classmethod
def _test_finalize(cls, conn):
class Foo(object):
result = [obj for obj in iter(conn.recv, 'STOP')]
self.assertEqual(result, ['a', 'b', 'd10', 'd03', 'd02', 'd01', 'e'])
+ def test_thread_safety(self):
+ # bpo-24484: _run_finalizers() should be thread-safe
+ def cb():
+ pass
+
+ class Foo(object):
+ def __init__(self):
+ self.ref = self # create reference cycle
+ # insert finalizer at random key
+ util.Finalize(self, cb, exitpriority=random.randint(1, 100))
+
+ finish = False
+ exc = []
+
+ def run_finalizers():
+ while not finish:
+ time.sleep(random.random() * 1e-1)
+ try:
+ # A GC run will eventually happen during this,
+ # collecting stale Foo's and mutating the registry
+ util._run_finalizers()
+ except Exception as e:
+ exc.append(e)
+
+ def make_finalizers():
+ d = {}
+ while not finish:
+ try:
+ # Old Foo's get gradually replaced and later
+ # collected by the GC (because of the cyclic ref)
+ d[random.getrandbits(5)] = {Foo() for i in range(10)}
+ except Exception as e:
+ exc.append(e)
+ d.clear()
+
+ old_interval = sys.getcheckinterval()
+ old_threshold = gc.get_threshold()
+ try:
+ sys.setcheckinterval(10)
+ gc.set_threshold(5, 5, 5)
+ threads = [threading.Thread(target=run_finalizers),
+ threading.Thread(target=make_finalizers)]
+ with test_support.start_threads(threads):
+ time.sleep(4.0) # Wait a bit to trigger race condition
+ finish = True
+ if exc:
+ raise exc[0]
+ finally:
+ sys.setcheckinterval(old_interval)
+ gc.set_threshold(*old_threshold)
+ gc.collect() # Collect remaining Foo's
+
+
#
# Test that from ... import * works for each module
#