Assert on specialisation stability for subclasses

Author: johng

Lib/test/test_opcache.py

@@ -38,6 +38,47 @@ def assert_no_opcode(self, f, opname):
         opnames = {instruction.opname for instruction in instructions}
         self.assertNotIn(opname, opnames)
 
+    def adaptive_counters(self, f):
+        """Map each specialized instruction in f to its adaptive counter."""
+        counters = {}
+        for instruction in dis.get_instructions(f, adaptive=True):
+            if instruction.opname == instruction.baseopname:
+                continue
+            if instruction.baseopname in ("RESUME", "JUMP_BACKWARD"):
+                continue
+            cache = {name: data for name, _, data in instruction.cache_info}
+            if "counter" in cache:
+                counters[instruction.offset] = (instruction.opname,
+                                                cache["counter"])
+        return counters
+
+    def assert_specialization_stable(self, f, *args, calls=10):
+        """Assert that no specialized instruction in f misses its guard.
+
+        A specialized instruction whose guard holds leaves its adaptive
+        counter untouched.  One whose guard fails on every execution
+        (the specializer accepted a site its guard then rejects) advances
+        the counter on each miss, so comparing the counters before and
+        after a few more calls detects the mismatch without needing a
+        pystats build.
+
+        f must start from a fresh code object (see reset_code()): a nested
+        function reuses its code object across calls of the enclosing test,
+        so a specialization left over from a previous run with different
+        operands would show up here as a miss.
+        """
+        before = self.adaptive_counters(f)
+        self.assertTrue(before, f"{f.__qualname__} has no specialized "
+                                "instructions")
+        for _ in range(calls):
+            f(*args)
+        after = self.adaptive_counters(f)
+        moved = [f"{op} at offset {off}: counter {before[off][1]} -> {c}"
+                 for off, (op, c) in after.items() if before[off] != (op, c)]
+        self.assertEqual(moved, [],
+                         f"specialized instructions in {f.__qualname__} "
+                         f"missed their guard during {calls} calls")
+
 
 class TestLoadSuperAttrCache(unittest.TestCase):
     def test_descriptor_not_double_executed_on_spec_fail(self):
@@ -2156,6 +2197,7 @@ def list_append(l):
             for _ in r:
                 l.append(1)
 
+        reset_code(list_append)
         list_append([])
         self.assert_specialized(list_append, "CALL_LIST_APPEND")
         self.assert_no_opcode(list_append, "CALL_METHOD_DESCRIPTOR_O")
@@ -2166,10 +2208,47 @@ def my_list_append(l):
                 l.append(1)
 
         class MyList(list): pass
-        my_list_append(MyList())
+        my_list = MyList()
+        reset_code(my_list_append)
+        my_list_append(my_list)
         self.assert_specialized(my_list_append, "CALL_METHOD_DESCRIPTOR_O")
         self.assert_no_opcode(my_list_append, "CALL_LIST_APPEND")
         self.assert_no_opcode(my_list_append, "CALL")
+        self.assert_specialization_stable(my_list_append, my_list)
+
+    @cpython_only
+    @requires_specialization
+    @requires_jit_disabled   # JIT-compiled code stops moving tier-1 counters
+    def test_call_method_descriptor_subclass_instance(self):
+        # A C method inherited from a built-in type is specialized for
+        # subclass instances too; each guard must accept the subtype
+        # rather than miss on every call.  One case per guard, plus the
+        # unbound form where the receiver is the first argument.
+        class MyStr(str): pass
+        Point = collections.namedtuple("Point", "x y")
+        counts = collections.defaultdict(int, {1: 2})
+
+        def noargs(s): return s.upper()
+        def o(p): return p.count(1)
+        def fast(d): return d.get(1)
+        def fast_with_keywords(s): return s.split()
+        def unbound(d): return dict.get(d, 1)
+
+        cases = [
+            (noargs, MyStr("abc"), "CALL_METHOD_DESCRIPTOR_NOARGS"),
+            (o, Point(1, 2), "CALL_METHOD_DESCRIPTOR_O"),
+            (fast, counts, "CALL_METHOD_DESCRIPTOR_FAST"),
+            (fast_with_keywords, MyStr("a b"),
+             "CALL_METHOD_DESCRIPTOR_FAST_WITH_KEYWORDS"),
+            (unbound, counts, "CALL_METHOD_DESCRIPTOR_FAST"),
+        ]
+        for f, obj, opname in cases:
+            with self.subTest(f=f.__name__, obj=type(obj).__name__):
+                reset_code(f)
+                for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD):
+                    f(obj)
+                self.assert_specialized(f, opname)
+                self.assert_specialization_stable(f, obj)
 
     @cpython_only
     @requires_specialization