From 0d5bd4aba77f28d088d447adab405b928bcda060 Mon Sep 17 00:00:00 2001 From: Shahar Naveh <50263213+ShaharNaveh@users.noreply.github.com> Date: Thu, 2 Apr 2026 18:33:34 +0200 Subject: [PATCH] Update `test_opcache.py` from 3.14.3 (#7553) --- Lib/test/test_opcache.py | 866 +++++++++++++++++++++++++++++++++++---- 1 file changed, 782 insertions(+), 84 deletions(-) diff --git a/Lib/test/test_opcache.py b/Lib/test/test_opcache.py index aeb4de6ca..2e50c1c62 100644 --- a/Lib/test/test_opcache.py +++ b/Lib/test/test_opcache.py @@ -4,7 +4,9 @@ import dis import threading import types import unittest -from test.support import threading_helper, check_impl_detail, requires_specialization +from test.support import (threading_helper, check_impl_detail, + requires_specialization, requires_specialization_ft, + cpython_only, requires_jit_disabled, reset_code) from test.support.import_helper import import_module # Skip this module on other interpreters, it is cpython specific: @@ -14,18 +16,14 @@ if check_impl_detail(cpython=False): _testinternalcapi = import_module("_testinternalcapi") -def disabling_optimizer(func): - def wrapper(*args, **kwargs): - if not hasattr(_testinternalcapi, "get_optimizer"): - return func(*args, **kwargs) - old_opt = _testinternalcapi.get_optimizer() - _testinternalcapi.set_optimizer(None) - try: - return func(*args, **kwargs) - finally: - _testinternalcapi.set_optimizer(old_opt) - - return wrapper +def have_dict_key_versions(): + # max version value that can be stored in the load global cache. This is + # determined by the type of module_keys_version and builtin_keys_version + # in _PyLoadGlobalCache, uint16_t. + max_version = 1<<16 + # use a wide safety margin (use only half of what's available) + limit = max_version // 2 + return _testinternalcapi.get_next_dict_keys_version() < limit class TestBase(unittest.TestCase): @@ -34,6 +32,11 @@ class TestBase(unittest.TestCase): opnames = {instruction.opname for instruction in instructions} self.assertIn(opname, opnames) + def assert_no_opcode(self, f, opname): + instructions = dis.get_instructions(f, adaptive=True) + opnames = {instruction.opname for instruction in instructions} + self.assertNotIn(opname, opnames) + class TestLoadSuperAttrCache(unittest.TestCase): def test_descriptor_not_double_executed_on_spec_fail(self): @@ -52,7 +55,8 @@ class TestLoadSuperAttrCache(unittest.TestCase): d = D() - self.assertEqual(d.f(), 1) # warmup + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD - 1): + self.assertEqual(d.f(), 1) # warmup calls.clear() self.assertEqual(d.f(), 1) # try to specialize self.assertEqual(calls, [(d, D)]) @@ -72,7 +76,7 @@ class TestLoadAttrCache(unittest.TestCase): return o.x o = C() - for i in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): assert f(o) == 1 Descriptor.__get__ = lambda self, instance, value: 2 @@ -99,13 +103,13 @@ class TestLoadAttrCache(unittest.TestCase): def f(): return Class.attribute - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): self.assertTrue(f()) Descriptor.__get__ = __get__ Descriptor.__set__ = __set__ - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_COOLDOWN): self.assertFalse(f()) def test_metaclass_descriptor_shadows_class_attribute(self): @@ -120,7 +124,7 @@ class TestLoadAttrCache(unittest.TestCase): def f(): return Class.attribute - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): self.assertTrue(f()) def test_metaclass_set_descriptor_after_optimization(self): @@ -137,12 +141,12 @@ class TestLoadAttrCache(unittest.TestCase): def f(): return Class.attribute - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): self.assertTrue(f()) Metaclass.attribute = attribute - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_COOLDOWN): self.assertFalse(f()) def test_metaclass_del_descriptor_after_optimization(self): @@ -157,12 +161,12 @@ class TestLoadAttrCache(unittest.TestCase): def f(): return Class.attribute - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): self.assertTrue(f()) del Metaclass.attribute - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_COOLDOWN): self.assertFalse(f()) def test_type_descriptor_shadows_attribute_method(self): @@ -172,7 +176,7 @@ class TestLoadAttrCache(unittest.TestCase): def f(): return Class.mro - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): self.assertIsNone(f()) def test_type_descriptor_shadows_attribute_member(self): @@ -182,7 +186,7 @@ class TestLoadAttrCache(unittest.TestCase): def f(): return Class.__base__ - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): self.assertIs(f(), object) def test_type_descriptor_shadows_attribute_getset(self): @@ -192,7 +196,7 @@ class TestLoadAttrCache(unittest.TestCase): def f(): return Class.__name__ - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): self.assertEqual(f(), "Class") def test_metaclass_getattribute(self): @@ -206,7 +210,7 @@ class TestLoadAttrCache(unittest.TestCase): def f(): return Class.attribute - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): self.assertTrue(f()) def test_metaclass_swap(self): @@ -226,12 +230,12 @@ class TestLoadAttrCache(unittest.TestCase): def f(): return Class.attribute - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): self.assertTrue(f()) Class.__class__ = NewMetaclass - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_COOLDOWN): self.assertFalse(f()) def test_load_shadowing_slot_should_raise_type_error(self): @@ -248,7 +252,7 @@ class TestLoadAttrCache(unittest.TestCase): o = Sneaky() o.shadowed = 42 - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): with self.assertRaises(TypeError): f(o) @@ -265,7 +269,7 @@ class TestLoadAttrCache(unittest.TestCase): o = Sneaky() - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): with self.assertRaises(TypeError): f(o) @@ -281,7 +285,7 @@ class TestLoadAttrCache(unittest.TestCase): o = Sneaky() - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): with self.assertRaises(TypeError): f(o) @@ -297,7 +301,7 @@ class TestLoadAttrCache(unittest.TestCase): o = Sneaky() - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): with self.assertRaises(TypeError): f(o) @@ -325,13 +329,13 @@ class TestLoadMethodCache(unittest.TestCase): def f(): return instance.attribute() - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): self.assertTrue(f()) Descriptor.__get__ = __get__ Descriptor.__set__ = __set__ - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_COOLDOWN): self.assertFalse(f()) def test_metaclass_descriptor_added_after_optimization(self): @@ -354,13 +358,13 @@ class TestLoadMethodCache(unittest.TestCase): def f(): return Class.attribute() - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): self.assertTrue(f()) Descriptor.__get__ = __get__ Descriptor.__set__ = __set__ - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_COOLDOWN): self.assertFalse(f()) def test_metaclass_descriptor_shadows_class_attribute(self): @@ -376,7 +380,7 @@ class TestLoadMethodCache(unittest.TestCase): def f(): return Class.attribute() - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): self.assertTrue(f()) def test_metaclass_set_descriptor_after_optimization(self): @@ -394,12 +398,12 @@ class TestLoadMethodCache(unittest.TestCase): def f(): return Class.attribute() - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): self.assertTrue(f()) Metaclass.attribute = attribute - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_COOLDOWN): self.assertFalse(f()) def test_metaclass_del_descriptor_after_optimization(self): @@ -415,12 +419,12 @@ class TestLoadMethodCache(unittest.TestCase): def f(): return Class.attribute() - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): self.assertTrue(f()) del Metaclass.attribute - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_COOLDOWN): self.assertFalse(f()) def test_type_descriptor_shadows_attribute_method(self): @@ -431,7 +435,7 @@ class TestLoadMethodCache(unittest.TestCase): def f(): return Class.mro() - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): self.assertEqual(f(), ["Spam", "eggs"]) def test_type_descriptor_shadows_attribute_member(self): @@ -442,7 +446,7 @@ class TestLoadMethodCache(unittest.TestCase): def f(): return Class.__base__() - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): self.assertNotEqual(f(), "Spam") def test_metaclass_getattribute(self): @@ -457,7 +461,7 @@ class TestLoadMethodCache(unittest.TestCase): def f(): return Class.attribute() - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): self.assertTrue(f()) def test_metaclass_swap(self): @@ -477,22 +481,27 @@ class TestLoadMethodCache(unittest.TestCase): def f(): return Class.attribute() - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): self.assertTrue(f()) Class.__class__ = NewMetaclass - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_COOLDOWN): self.assertFalse(f()) +class InitTakesArg: + def __init__(self, arg): + self.arg = arg + + class TestCallCache(TestBase): def test_too_many_defaults_0(self): def f(): pass f.__defaults__ = (None,) - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): f() def test_too_many_defaults_1(self): @@ -500,7 +509,7 @@ class TestCallCache(TestBase): pass f.__defaults__ = (None, None) - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): f(None) f() @@ -509,13 +518,13 @@ class TestCallCache(TestBase): pass f.__defaults__ = (None, None, None) - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): f(None, None) f(None) f() - @disabling_optimizer - @requires_specialization + @requires_jit_disabled + @requires_specialization_ft def test_assign_init_code(self): class MyClass: def __init__(self): @@ -525,30 +534,67 @@ class TestCallCache(TestBase): return MyClass() # Trigger specialization - for _ in range(1025): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): instantiate() self.assert_specialized(instantiate, "CALL_ALLOC_AND_ENTER_INIT") def count_args(self, *args): self.num_args = len(args) - # Set MyClass.__init__.__code__ to a code object that is incompatible - # (uses varargs) with the current specialization + # Set MyClass.__init__.__code__ to a code object that uses different + # args MyClass.__init__.__code__ = count_args.__code__ instantiate() + @requires_jit_disabled + @requires_specialization_ft + def test_push_init_frame_fails(self): + def instantiate(): + return InitTakesArg() + + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): + with self.assertRaises(TypeError): + instantiate() + self.assert_specialized(instantiate, "CALL_ALLOC_AND_ENTER_INIT") + + with self.assertRaises(TypeError): + instantiate() + + def test_recursion_check_for_general_calls(self): + def test(default=None): + return test() + + with self.assertRaises(RecursionError): + test() + + def test_dont_specialize_custom_vectorcall(self): + def f(): + raise Exception("no way") + + _testinternalcapi.set_vectorcall_nop(f) + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): + f() + + +def make_deferred_ref_count_obj(): + """Create an object that uses deferred reference counting. + + Only objects that use deferred refence counting may be stored in inline + caches in free-threaded builds. This constructs a new class named Foo, + which uses deferred reference counting. + """ + return type("Foo", (object,), {}) + @threading_helper.requires_working_threading() -@requires_specialization class TestRacesDoNotCrash(TestBase): # Careful with these. Bigger numbers have a higher chance of catching bugs, # but you can also burn through a *ton* of type/dict/function versions: ITEMS = 1000 LOOPS = 4 - WARMUPS = 2 WRITERS = 2 - @disabling_optimizer + @requires_jit_disabled def assert_races_do_not_crash( self, opname, get_items, read, write, *, check_items=False ): @@ -558,11 +604,11 @@ class TestRacesDoNotCrash(TestBase): # Reset: if check_items: for item in items: - item.__code__ = item.__code__.replace() + reset_code(item) else: - read.__code__ = read.__code__.replace() + reset_code(read) # Specialize: - for _ in range(self.WARMUPS): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): read(items) if check_items: for item in items: @@ -581,6 +627,7 @@ class TestRacesDoNotCrash(TestBase): for writer in writers: writer.join() + @requires_specialization_ft def test_binary_subscr_getitem(self): def get_items(): class C: @@ -607,9 +654,10 @@ class TestRacesDoNotCrash(TestBase): pass type(item).__getitem__ = lambda self, item: None - opname = "BINARY_SUBSCR_GETITEM" + opname = "BINARY_OP_SUBSCR_GETITEM" self.assert_races_do_not_crash(opname, get_items, read, write) + @requires_specialization_ft def test_binary_subscr_list_int(self): def get_items(): items = [] @@ -630,9 +678,10 @@ class TestRacesDoNotCrash(TestBase): item.clear() item.append(None) - opname = "BINARY_SUBSCR_LIST_INT" + opname = "BINARY_OP_SUBSCR_LIST_INT" self.assert_races_do_not_crash(opname, get_items, read, write) + @requires_specialization def test_for_iter_gen(self): def get_items(): def g(): @@ -664,6 +713,7 @@ class TestRacesDoNotCrash(TestBase): opname = "FOR_ITER_GEN" self.assert_races_do_not_crash(opname, get_items, read, write) + @requires_specialization def test_for_iter_list(self): def get_items(): items = [] @@ -685,10 +735,11 @@ class TestRacesDoNotCrash(TestBase): opname = "FOR_ITER_LIST" self.assert_races_do_not_crash(opname, get_items, read, write) + @requires_specialization_ft def test_load_attr_class(self): def get_items(): class C: - a = object() + a = make_deferred_ref_count_obj() items = [] for _ in range(self.ITEMS): @@ -709,11 +760,45 @@ class TestRacesDoNotCrash(TestBase): del item.a except AttributeError: pass - item.a = object() + item.a = make_deferred_ref_count_obj() opname = "LOAD_ATTR_CLASS" self.assert_races_do_not_crash(opname, get_items, read, write) + @requires_specialization_ft + def test_load_attr_class_with_metaclass_check(self): + def get_items(): + class Meta(type): + pass + + class C(metaclass=Meta): + a = make_deferred_ref_count_obj() + + items = [] + for _ in range(self.ITEMS): + item = C + items.append(item) + return items + + def read(items): + for item in items: + try: + item.a + except AttributeError: + pass + + def write(items): + for item in items: + try: + del item.a + except AttributeError: + pass + item.a = make_deferred_ref_count_obj() + + opname = "LOAD_ATTR_CLASS_WITH_METACLASS_CHECK" + self.assert_races_do_not_crash(opname, get_items, read, write) + + @requires_specialization_ft def test_load_attr_getattribute_overridden(self): def get_items(): class C: @@ -743,6 +828,7 @@ class TestRacesDoNotCrash(TestBase): opname = "LOAD_ATTR_GETATTRIBUTE_OVERRIDDEN" self.assert_races_do_not_crash(opname, get_items, read, write) + @requires_specialization_ft def test_load_attr_instance_value(self): def get_items(): class C: @@ -766,6 +852,7 @@ class TestRacesDoNotCrash(TestBase): opname = "LOAD_ATTR_INSTANCE_VALUE" self.assert_races_do_not_crash(opname, get_items, read, write) + @requires_specialization_ft def test_load_attr_method_lazy_dict(self): def get_items(): class C(Exception): @@ -795,6 +882,7 @@ class TestRacesDoNotCrash(TestBase): opname = "LOAD_ATTR_METHOD_LAZY_DICT" self.assert_races_do_not_crash(opname, get_items, read, write) + @requires_specialization_ft def test_load_attr_method_no_dict(self): def get_items(): class C: @@ -825,6 +913,7 @@ class TestRacesDoNotCrash(TestBase): opname = "LOAD_ATTR_METHOD_NO_DICT" self.assert_races_do_not_crash(opname, get_items, read, write) + @requires_specialization_ft def test_load_attr_method_with_values(self): def get_items(): class C: @@ -854,6 +943,7 @@ class TestRacesDoNotCrash(TestBase): opname = "LOAD_ATTR_METHOD_WITH_VALUES" self.assert_races_do_not_crash(opname, get_items, read, write) + @requires_specialization_ft def test_load_attr_module(self): def get_items(): items = [] @@ -878,6 +968,7 @@ class TestRacesDoNotCrash(TestBase): opname = "LOAD_ATTR_MODULE" self.assert_races_do_not_crash(opname, get_items, read, write) + @requires_specialization_ft def test_load_attr_property(self): def get_items(): class C: @@ -907,6 +998,34 @@ class TestRacesDoNotCrash(TestBase): opname = "LOAD_ATTR_PROPERTY" self.assert_races_do_not_crash(opname, get_items, read, write) + @requires_specialization_ft + def test_load_attr_slot(self): + def get_items(): + class C: + __slots__ = ["a", "b"] + + items = [] + for i in range(self.ITEMS): + item = C() + item.a = i + item.b = i + self.ITEMS + items.append(item) + return items + + def read(items): + for item in items: + item.a + item.b + + def write(items): + for item in items: + item.a = 100 + item.b = 200 + + opname = "LOAD_ATTR_SLOT" + self.assert_races_do_not_crash(opname, get_items, read, write) + + @requires_specialization_ft def test_load_attr_with_hint(self): def get_items(): class C: @@ -917,7 +1036,7 @@ class TestRacesDoNotCrash(TestBase): item = C() item.a = None # Resize into a combined unicode dict: - for i in range(29): + for i in range(_testinternalcapi.SHARED_KEYS_MAX_SIZE - 1): setattr(item, f"_{i}", None) items.append(item) return items @@ -933,7 +1052,10 @@ class TestRacesDoNotCrash(TestBase): opname = "LOAD_ATTR_WITH_HINT" self.assert_races_do_not_crash(opname, get_items, read, write) + @requires_specialization_ft def test_load_global_module(self): + if not have_dict_key_versions(): + raise unittest.SkipTest("Low on dict key versions") def get_items(): items = [] for _ in range(self.ITEMS): @@ -954,6 +1076,7 @@ class TestRacesDoNotCrash(TestBase): opname, get_items, read, write, check_items=True ) + @requires_specialization def test_store_attr_instance_value(self): def get_items(): class C: @@ -976,6 +1099,7 @@ class TestRacesDoNotCrash(TestBase): opname = "STORE_ATTR_INSTANCE_VALUE" self.assert_races_do_not_crash(opname, get_items, read, write) + @requires_specialization def test_store_attr_with_hint(self): def get_items(): class C: @@ -985,7 +1109,7 @@ class TestRacesDoNotCrash(TestBase): for _ in range(self.ITEMS): item = C() # Resize into a combined unicode dict: - for i in range(29): + for i in range(_testinternalcapi.SHARED_KEYS_MAX_SIZE - 1): setattr(item, f"_{i}", None) items.append(item) return items @@ -1001,6 +1125,7 @@ class TestRacesDoNotCrash(TestBase): opname = "STORE_ATTR_WITH_HINT" self.assert_races_do_not_crash(opname, get_items, read, write) + @requires_specialization_ft def test_store_subscr_list_int(self): def get_items(): items = [] @@ -1024,6 +1149,7 @@ class TestRacesDoNotCrash(TestBase): opname = "STORE_SUBSCR_LIST_INT" self.assert_races_do_not_crash(opname, get_items, read, write) + @requires_specialization_ft def test_unpack_sequence_list(self): def get_items(): items = [] @@ -1079,7 +1205,7 @@ class TestInstanceDict(unittest.TestCase): c.a = 1 c.b = 2 c.__dict__ - for _ in range(100): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): c.a self.assertEqual( _testinternalcapi.get_object_dict_values(c), @@ -1091,7 +1217,7 @@ class TestInstanceDict(unittest.TestCase): c.a = 1 c.b = 2 d = c.__dict__ - for _ in range(100): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): c.a self.assertIs(c.__dict__, d) @@ -1100,7 +1226,7 @@ class TestInstanceDict(unittest.TestCase): c.a = 1 c.b = 2 c2 = copy.copy(c) - for _ in range(100): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): c.a c2.a self.assertEqual( @@ -1112,7 +1238,7 @@ class TestInstanceDict(unittest.TestCase): (1, 2, '') ) c3 = copy.deepcopy(c) - for _ in range(100): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): c.a c3.a self.assertEqual( @@ -1126,7 +1252,7 @@ class TestInstanceDict(unittest.TestCase): c.a = 1 c.b = 2 c2 = pickle.loads(pickle.dumps(c)) - for _ in range(100): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): c.a c2.a self.assertEqual( @@ -1144,7 +1270,7 @@ class TestInstanceDict(unittest.TestCase): c.a = 1 c.b = 2 c.__dict__ = D(c.__dict__) - for _ in range(100): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): c.a self.assertIs( _testinternalcapi.get_object_dict_values(c), @@ -1155,23 +1281,595 @@ class TestInstanceDict(unittest.TestCase): {'a':1, 'b':2} ) - def test_store_attr_with_hint(self): - # gh-133441: Regression test for STORE_ATTR_WITH_HINT bytecode - class Node: + def test_125868(self): + + def make_special_dict(): + """Create a dictionary an object with a this table: + index | key | value + ----- | --- | ----- + 0 | 'b' | 'value' + 1 | 'b' | NULL + """ + class A: + pass + a = A() + a.a = 1 + a.b = 2 + d = a.__dict__.copy() + del d['a'] + del d['b'] + d['b'] = "value" + return d + + class NoInlineAorB: + pass + for i in range(ord('c'), ord('z')): + setattr(NoInlineAorB(), chr(i), i) + + c = NoInlineAorB() + c.a = 0 + c.b = 1 + self.assertFalse(_testinternalcapi.has_inline_values(c)) + + def f(o, n): + for i in range(n): + o.b = i + # Prime f to store to dict slot 1 + f(c, _testinternalcapi.SPECIALIZATION_THRESHOLD) + + test_obj = NoInlineAorB() + test_obj.__dict__ = make_special_dict() + self.assertEqual(test_obj.b, "value") + + #This should set x.b = 0 + f(test_obj, 1) + self.assertEqual(test_obj.b, 0) + + +class TestSpecializer(TestBase): + + @cpython_only + @requires_specialization_ft + def test_binary_op(self): + def binary_op_add_int(): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): + a, b = 1, 2 + c = a + b + self.assertEqual(c, 3) + + binary_op_add_int() + self.assert_specialized(binary_op_add_int, "BINARY_OP_ADD_INT") + self.assert_no_opcode(binary_op_add_int, "BINARY_OP") + + def binary_op_add_unicode(): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): + a, b = "foo", "bar" + c = a + b + self.assertEqual(c, "foobar") + + binary_op_add_unicode() + self.assert_specialized(binary_op_add_unicode, "BINARY_OP_ADD_UNICODE") + self.assert_no_opcode(binary_op_add_unicode, "BINARY_OP") + + def binary_op_add_extend(): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): + a, b = 6, 3.0 + c = a + b + self.assertEqual(c, 9.0) + c = b + a + self.assertEqual(c, 9.0) + c = a - b + self.assertEqual(c, 3.0) + c = b - a + self.assertEqual(c, -3.0) + c = a * b + self.assertEqual(c, 18.0) + c = b * a + self.assertEqual(c, 18.0) + c = a / b + self.assertEqual(c, 2.0) + c = b / a + self.assertEqual(c, 0.5) + + binary_op_add_extend() + self.assert_specialized(binary_op_add_extend, "BINARY_OP_EXTEND") + self.assert_no_opcode(binary_op_add_extend, "BINARY_OP") + + def binary_op_zero_division(): + def compactlong_lhs(arg): + 42 / arg + def float_lhs(arg): + 42.0 / arg + + with self.assertRaises(ZeroDivisionError): + compactlong_lhs(0) + with self.assertRaises(ZeroDivisionError): + compactlong_lhs(0.0) + with self.assertRaises(ZeroDivisionError): + float_lhs(0.0) + with self.assertRaises(ZeroDivisionError): + float_lhs(0) + + self.assert_no_opcode(compactlong_lhs, "BINARY_OP_EXTEND") + self.assert_no_opcode(float_lhs, "BINARY_OP_EXTEND") + + binary_op_zero_division() + + def binary_op_nan(): + def compactlong_lhs(arg): + return ( + 42 + arg, + 42 - arg, + 42 * arg, + 42 / arg, + ) + def compactlong_rhs(arg): + return ( + arg + 42, + arg - 42, + arg * 2, + arg / 42, + ) + nan = float('nan') + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): + self.assertEqual(compactlong_lhs(1.0), (43.0, 41.0, 42.0, 42.0)) + for _ in range(_testinternalcapi.SPECIALIZATION_COOLDOWN): + self.assertTrue(all(filter(lambda x: x is nan, compactlong_lhs(nan)))) + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): + self.assertEqual(compactlong_rhs(42.0), (84.0, 0.0, 84.0, 1.0)) + for _ in range(_testinternalcapi.SPECIALIZATION_COOLDOWN): + self.assertTrue(all(filter(lambda x: x is nan, compactlong_rhs(nan)))) + + self.assert_no_opcode(compactlong_lhs, "BINARY_OP_EXTEND") + self.assert_no_opcode(compactlong_rhs, "BINARY_OP_EXTEND") + + binary_op_nan() + + def binary_op_bitwise_extend(): + for _ in range(100): + a, b = 2, 7 + x = a | b + self.assertEqual(x, 7) + y = a & b + self.assertEqual(y, 2) + z = a ^ b + self.assertEqual(z, 5) + a, b = 3, 9 + a |= b + self.assertEqual(a, 11) + a, b = 11, 9 + a &= b + self.assertEqual(a, 9) + a, b = 3, 9 + a ^= b + self.assertEqual(a, 10) + + binary_op_bitwise_extend() + self.assert_specialized(binary_op_bitwise_extend, "BINARY_OP_EXTEND") + self.assert_no_opcode(binary_op_bitwise_extend, "BINARY_OP") + + @cpython_only + @requires_specialization_ft + def test_load_super_attr(self): + """Ensure that LOAD_SUPER_ATTR is specialized as expected.""" + + class A: def __init__(self): - self.parents = {} + meth = super().__init__ + super().__init__() - def __setstate__(self, data_dict): - self.__dict__ = data_dict - self.parents = {} + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): + A() - class Dict(dict): + self.assert_specialized(A.__init__, "LOAD_SUPER_ATTR_ATTR") + self.assert_specialized(A.__init__, "LOAD_SUPER_ATTR_METHOD") + self.assert_no_opcode(A.__init__, "LOAD_SUPER_ATTR") + + # Temporarily replace super() with something else. + real_super = super + + def fake_super(): + def init(self): + pass + + return init + + # Force unspecialize + globals()['super'] = fake_super + try: + # Should be unspecialized after enough calls. + for _ in range(_testinternalcapi.SPECIALIZATION_COOLDOWN): + A() + finally: + globals()['super'] = real_super + + # Ensure the specialized instructions are not present + self.assert_no_opcode(A.__init__, "LOAD_SUPER_ATTR_ATTR") + self.assert_no_opcode(A.__init__, "LOAD_SUPER_ATTR_METHOD") + + @cpython_only + @requires_specialization_ft + def test_contain_op(self): + def contains_op_dict(): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): + a, b = 1, {1: 2, 2: 5} + self.assertTrue(a in b) + self.assertFalse(3 in b) + + contains_op_dict() + self.assert_specialized(contains_op_dict, "CONTAINS_OP_DICT") + self.assert_no_opcode(contains_op_dict, "CONTAINS_OP") + + def contains_op_set(): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): + a, b = 1, {1, 2} + self.assertTrue(a in b) + self.assertFalse(3 in b) + + contains_op_set() + self.assert_specialized(contains_op_set, "CONTAINS_OP_SET") + self.assert_no_opcode(contains_op_set, "CONTAINS_OP") + + @cpython_only + @requires_specialization_ft + def test_send_with(self): + def run_async(coro): + while True: + try: + coro.send(None) + except StopIteration: + break + + class CM: + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + pass + + async def send_with(): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): + async with CM(): + x = 1 + + run_async(send_with()) + # Note there are still unspecialized "SEND" opcodes in the + # cleanup paths of the 'with' statement. + self.assert_specialized(send_with, "SEND_GEN") + + @cpython_only + @requires_specialization_ft + def test_send_yield_from(self): + def g(): + yield None + + def send_yield_from(): + yield from g() + + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): + list(send_yield_from()) + + self.assert_specialized(send_yield_from, "SEND_GEN") + self.assert_no_opcode(send_yield_from, "SEND") + + @cpython_only + @requires_specialization_ft + def test_store_attr_slot(self): + class C: + __slots__ = ['x'] + + def set_slot(n): + c = C() + for i in range(n): + c.x = i + + set_slot(_testinternalcapi.SPECIALIZATION_THRESHOLD) + + self.assert_specialized(set_slot, "STORE_ATTR_SLOT") + self.assert_no_opcode(set_slot, "STORE_ATTR") + + # Adding a property for 'x' should unspecialize it. + C.x = property(lambda self: None, lambda self, x: None) + set_slot(_testinternalcapi.SPECIALIZATION_COOLDOWN) + self.assert_no_opcode(set_slot, "STORE_ATTR_SLOT") + + @cpython_only + @requires_specialization_ft + def test_store_attr_instance_value(self): + class C: pass - obj = Node() - obj.__setstate__({'parents': {}}) - obj.__setstate__({'parents': {}}) - obj.__setstate__(Dict({'parents': {}})) + @reset_code + def set_value(n): + c = C() + for i in range(n): + c.x = i + + set_value(_testinternalcapi.SPECIALIZATION_THRESHOLD) + + self.assert_specialized(set_value, "STORE_ATTR_INSTANCE_VALUE") + self.assert_no_opcode(set_value, "STORE_ATTR") + + # Adding a property for 'x' should unspecialize it. + C.x = property(lambda self: None, lambda self, x: None) + set_value(_testinternalcapi.SPECIALIZATION_COOLDOWN) + self.assert_no_opcode(set_value, "STORE_ATTR_INSTANCE_VALUE") + + @cpython_only + @requires_specialization_ft + def test_store_attr_with_hint(self): + class C: + pass + + c = C() + for i in range(_testinternalcapi.SHARED_KEYS_MAX_SIZE - 1): + setattr(c, f"_{i}", None) + + @reset_code + def set_value(n): + for i in range(n): + c.x = i + + set_value(_testinternalcapi.SPECIALIZATION_THRESHOLD) + + self.assert_specialized(set_value, "STORE_ATTR_WITH_HINT") + self.assert_no_opcode(set_value, "STORE_ATTR") + + # Adding a property for 'x' should unspecialize it. + C.x = property(lambda self: None, lambda self, x: None) + set_value(_testinternalcapi.SPECIALIZATION_COOLDOWN) + self.assert_no_opcode(set_value, "STORE_ATTR_WITH_HINT") + + @cpython_only + @requires_specialization_ft + def test_to_bool(self): + def to_bool_bool(): + true_cnt, false_cnt = 0, 0 + elems = [e % 2 == 0 for e in range(_testinternalcapi.SPECIALIZATION_THRESHOLD)] + for e in elems: + if e: + true_cnt += 1 + else: + false_cnt += 1 + d, m = divmod(_testinternalcapi.SPECIALIZATION_THRESHOLD, 2) + self.assertEqual(true_cnt, d + m) + self.assertEqual(false_cnt, d) + + to_bool_bool() + self.assert_specialized(to_bool_bool, "TO_BOOL_BOOL") + self.assert_no_opcode(to_bool_bool, "TO_BOOL") + + def to_bool_int(): + count = 0 + for i in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): + if i: + count += 1 + else: + count -= 1 + self.assertEqual(count, _testinternalcapi.SPECIALIZATION_THRESHOLD - 2) + + to_bool_int() + self.assert_specialized(to_bool_int, "TO_BOOL_INT") + self.assert_no_opcode(to_bool_int, "TO_BOOL") + + def to_bool_list(): + count = 0 + elems = list(range(_testinternalcapi.SPECIALIZATION_THRESHOLD)) + while elems: + count += elems.pop() + self.assertEqual(elems, []) + self.assertEqual(count, sum(range(_testinternalcapi.SPECIALIZATION_THRESHOLD))) + + to_bool_list() + self.assert_specialized(to_bool_list, "TO_BOOL_LIST") + self.assert_no_opcode(to_bool_list, "TO_BOOL") + + def to_bool_none(): + count = 0 + elems = [None] * _testinternalcapi.SPECIALIZATION_THRESHOLD + for e in elems: + if not e: + count += 1 + self.assertEqual(count, _testinternalcapi.SPECIALIZATION_THRESHOLD) + + to_bool_none() + self.assert_specialized(to_bool_none, "TO_BOOL_NONE") + self.assert_no_opcode(to_bool_none, "TO_BOOL") + + def to_bool_str(): + count = 0 + elems = [""] + ["foo"] * (_testinternalcapi.SPECIALIZATION_THRESHOLD - 1) + for e in elems: + if e: + count += 1 + self.assertEqual(count, _testinternalcapi.SPECIALIZATION_THRESHOLD - 1) + + to_bool_str() + self.assert_specialized(to_bool_str, "TO_BOOL_STR") + self.assert_no_opcode(to_bool_str, "TO_BOOL") + + @cpython_only + @requires_specialization_ft + def test_unpack_sequence(self): + def unpack_sequence_two_tuple(): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): + t = 1, 2 + a, b = t + self.assertEqual(a, 1) + self.assertEqual(b, 2) + + unpack_sequence_two_tuple() + self.assert_specialized(unpack_sequence_two_tuple, + "UNPACK_SEQUENCE_TWO_TUPLE") + self.assert_no_opcode(unpack_sequence_two_tuple, "UNPACK_SEQUENCE") + + def unpack_sequence_tuple(): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): + a, b, c, d = 1, 2, 3, 4 + self.assertEqual(a, 1) + self.assertEqual(b, 2) + self.assertEqual(c, 3) + self.assertEqual(d, 4) + + unpack_sequence_tuple() + self.assert_specialized(unpack_sequence_tuple, "UNPACK_SEQUENCE_TUPLE") + self.assert_no_opcode(unpack_sequence_tuple, "UNPACK_SEQUENCE") + + def unpack_sequence_list(): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): + a, b = [1, 2] + self.assertEqual(a, 1) + self.assertEqual(b, 2) + + unpack_sequence_list() + self.assert_specialized(unpack_sequence_list, "UNPACK_SEQUENCE_LIST") + self.assert_no_opcode(unpack_sequence_list, "UNPACK_SEQUENCE") + + @cpython_only + @requires_specialization_ft + def test_binary_subscr(self): + def binary_subscr_list_int(): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): + a = [1, 2, 3] + for idx, expected in enumerate(a): + self.assertEqual(a[idx], expected) + + binary_subscr_list_int() + self.assert_specialized(binary_subscr_list_int, + "BINARY_OP_SUBSCR_LIST_INT") + self.assert_no_opcode(binary_subscr_list_int, "BINARY_OP") + + def binary_subscr_tuple_int(): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): + a = (1, 2, 3) + for idx, expected in enumerate(a): + self.assertEqual(a[idx], expected) + + binary_subscr_tuple_int() + self.assert_specialized(binary_subscr_tuple_int, + "BINARY_OP_SUBSCR_TUPLE_INT") + self.assert_no_opcode(binary_subscr_tuple_int, "BINARY_OP") + + def binary_subscr_dict(): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): + a = {1: 2, 2: 3} + self.assertEqual(a[1], 2) + self.assertEqual(a[2], 3) + + binary_subscr_dict() + self.assert_specialized(binary_subscr_dict, "BINARY_OP_SUBSCR_DICT") + self.assert_no_opcode(binary_subscr_dict, "BINARY_OP") + + def binary_subscr_str_int(): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): + a = "foobar" + for idx, expected in enumerate(a): + self.assertEqual(a[idx], expected) + + binary_subscr_str_int() + self.assert_specialized(binary_subscr_str_int, "BINARY_OP_SUBSCR_STR_INT") + self.assert_no_opcode(binary_subscr_str_int, "BINARY_OP") + + def binary_subscr_getitems(): + class C: + def __init__(self, val): + self.val = val + def __getitem__(self, item): + return self.val + + items = [C(i) for i in range(_testinternalcapi.SPECIALIZATION_THRESHOLD)] + for i in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): + self.assertEqual(items[i][i], i) + + binary_subscr_getitems() + self.assert_specialized(binary_subscr_getitems, "BINARY_OP_SUBSCR_GETITEM") + self.assert_no_opcode(binary_subscr_getitems, "BINARY_OP") + + @cpython_only + @requires_specialization_ft + def test_compare_op(self): + def compare_op_int(): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): + a, b = 1, 2 + c = a == b + self.assertFalse(c) + + compare_op_int() + self.assert_specialized(compare_op_int, "COMPARE_OP_INT") + self.assert_no_opcode(compare_op_int, "COMPARE_OP") + + def compare_op_float(): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): + a, b = 1.0, 2.0 + c = a == b + self.assertFalse(c) + + compare_op_float() + self.assert_specialized(compare_op_float, "COMPARE_OP_FLOAT") + self.assert_no_opcode(compare_op_float, "COMPARE_OP") + + def compare_op_str(): + for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD): + a, b = "spam", "ham" + c = a == b + self.assertFalse(c) + + compare_op_str() + self.assert_specialized(compare_op_str, "COMPARE_OP_STR") + self.assert_no_opcode(compare_op_str, "COMPARE_OP") + + @cpython_only + @requires_specialization_ft + def test_load_const(self): + def load_const(): + def unused(): pass + # Currently, the empty tuple is immortal, and the otherwise + # unused nested function's code object is mortal. This test will + # have to use different values if either of that changes. + return () + + load_const() + self.assert_specialized(load_const, "LOAD_CONST_IMMORTAL") + self.assert_specialized(load_const, "LOAD_CONST_MORTAL") + self.assert_no_opcode(load_const, "LOAD_CONST") + + @cpython_only + @requires_specialization_ft + def test_for_iter(self): + L = list(range(10)) + def for_iter_list(): + for i in L: + self.assertIn(i, L) + + for_iter_list() + self.assert_specialized(for_iter_list, "FOR_ITER_LIST") + self.assert_no_opcode(for_iter_list, "FOR_ITER") + + t = tuple(range(10)) + def for_iter_tuple(): + for i in t: + self.assertIn(i, t) + + for_iter_tuple() + self.assert_specialized(for_iter_tuple, "FOR_ITER_TUPLE") + self.assert_no_opcode(for_iter_tuple, "FOR_ITER") + + r = range(10) + def for_iter_range(): + for i in r: + self.assertIn(i, r) + + for_iter_range() + self.assert_specialized(for_iter_range, "FOR_ITER_RANGE") + self.assert_no_opcode(for_iter_range, "FOR_ITER") + + def for_iter_generator(): + for i in (i for i in range(10)): + i + 1 + + for_iter_generator() + self.assert_specialized(for_iter_generator, "FOR_ITER_GEN") + self.assert_no_opcode(for_iter_generator, "FOR_ITER") if __name__ == "__main__":