mirror of
				https://github.com/python/cpython.git
				synced 2025-11-03 19:34:08 +00:00 
			
		
		
		
	This merges all `_CHECK_STACK_SPACE` uops in a trace into a single `_CHECK_STACK_SPACE_OPERAND` uop that checks whether there is enough stack space for all calls included in the entire trace.
		
			
				
	
	
		
			1247 lines
		
	
	
	
		
			41 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
			
		
		
	
	
			1247 lines
		
	
	
	
		
			41 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
import contextlib
 | 
						|
import sys
 | 
						|
import textwrap
 | 
						|
import unittest
 | 
						|
import gc
 | 
						|
import os
 | 
						|
 | 
						|
import _opcode
 | 
						|
import _testinternalcapi
 | 
						|
 | 
						|
from test.support import script_helper, requires_specialization
 | 
						|
 | 
						|
 | 
						|
@contextlib.contextmanager
 | 
						|
def temporary_optimizer(opt):
 | 
						|
    old_opt = _testinternalcapi.get_optimizer()
 | 
						|
    _testinternalcapi.set_optimizer(opt)
 | 
						|
    try:
 | 
						|
        yield
 | 
						|
    finally:
 | 
						|
        _testinternalcapi.set_optimizer(old_opt)
 | 
						|
 | 
						|
 | 
						|
@contextlib.contextmanager
 | 
						|
def clear_executors(func):
 | 
						|
    # Clear executors in func before and after running a block
 | 
						|
    func.__code__ = func.__code__.replace()
 | 
						|
    try:
 | 
						|
        yield
 | 
						|
    finally:
 | 
						|
        func.__code__ = func.__code__.replace()
 | 
						|
 | 
						|
 | 
						|
@requires_specialization
 | 
						|
class TestOptimizerAPI(unittest.TestCase):
 | 
						|
 | 
						|
    def test_new_counter_optimizer_dealloc(self):
 | 
						|
        # See gh-108727
 | 
						|
        def f():
 | 
						|
            _testinternalcapi.new_counter_optimizer()
 | 
						|
 | 
						|
        f()
 | 
						|
 | 
						|
    def test_get_set_optimizer(self):
 | 
						|
        old = _testinternalcapi.get_optimizer()
 | 
						|
        opt = _testinternalcapi.new_counter_optimizer()
 | 
						|
        try:
 | 
						|
            _testinternalcapi.set_optimizer(opt)
 | 
						|
            self.assertEqual(_testinternalcapi.get_optimizer(), opt)
 | 
						|
            _testinternalcapi.set_optimizer(None)
 | 
						|
            self.assertEqual(_testinternalcapi.get_optimizer(), None)
 | 
						|
        finally:
 | 
						|
            _testinternalcapi.set_optimizer(old)
 | 
						|
 | 
						|
 | 
						|
    def test_counter_optimizer(self):
 | 
						|
        # Generate a new function at each call
 | 
						|
        ns = {}
 | 
						|
        exec(textwrap.dedent("""
 | 
						|
            def loop():
 | 
						|
                for _ in range(1000):
 | 
						|
                    pass
 | 
						|
        """), ns, ns)
 | 
						|
        loop = ns['loop']
 | 
						|
 | 
						|
        for repeat in range(5):
 | 
						|
            opt = _testinternalcapi.new_counter_optimizer()
 | 
						|
            with temporary_optimizer(opt):
 | 
						|
                self.assertEqual(opt.get_count(), 0)
 | 
						|
                with clear_executors(loop):
 | 
						|
                    loop()
 | 
						|
                self.assertEqual(opt.get_count(), 1000)
 | 
						|
 | 
						|
    def test_long_loop(self):
 | 
						|
        "Check that we aren't confused by EXTENDED_ARG"
 | 
						|
 | 
						|
        # Generate a new function at each call
 | 
						|
        ns = {}
 | 
						|
        exec(textwrap.dedent("""
 | 
						|
            def nop():
 | 
						|
                pass
 | 
						|
 | 
						|
            def long_loop():
 | 
						|
                for _ in range(10):
 | 
						|
                    nop(); nop(); nop(); nop(); nop(); nop(); nop(); nop();
 | 
						|
                    nop(); nop(); nop(); nop(); nop(); nop(); nop(); nop();
 | 
						|
                    nop(); nop(); nop(); nop(); nop(); nop(); nop(); nop();
 | 
						|
                    nop(); nop(); nop(); nop(); nop(); nop(); nop(); nop();
 | 
						|
                    nop(); nop(); nop(); nop(); nop(); nop(); nop(); nop();
 | 
						|
                    nop(); nop(); nop(); nop(); nop(); nop(); nop(); nop();
 | 
						|
                    nop(); nop(); nop(); nop(); nop(); nop(); nop(); nop();
 | 
						|
        """), ns, ns)
 | 
						|
        long_loop = ns['long_loop']
 | 
						|
 | 
						|
        opt = _testinternalcapi.new_counter_optimizer()
 | 
						|
        with temporary_optimizer(opt):
 | 
						|
            self.assertEqual(opt.get_count(), 0)
 | 
						|
            long_loop()
 | 
						|
            self.assertEqual(opt.get_count(), 10)
 | 
						|
 | 
						|
    def test_code_restore_for_ENTER_EXECUTOR(self):
 | 
						|
        def testfunc(x):
 | 
						|
            i = 0
 | 
						|
            while i < x:
 | 
						|
                i += 1
 | 
						|
 | 
						|
        opt = _testinternalcapi.new_counter_optimizer()
 | 
						|
        with temporary_optimizer(opt):
 | 
						|
            testfunc(1000)
 | 
						|
            code, replace_code  = testfunc.__code__, testfunc.__code__.replace()
 | 
						|
            self.assertEqual(code, replace_code)
 | 
						|
            self.assertEqual(hash(code), hash(replace_code))
 | 
						|
 | 
						|
 | 
						|
def get_first_executor(func):
 | 
						|
    code = func.__code__
 | 
						|
    co_code = code.co_code
 | 
						|
    for i in range(0, len(co_code), 2):
 | 
						|
        try:
 | 
						|
            return _opcode.get_executor(code, i)
 | 
						|
        except ValueError:
 | 
						|
            pass
 | 
						|
    return None
 | 
						|
 | 
						|
 | 
						|
def iter_opnames(ex):
 | 
						|
    for item in ex:
 | 
						|
        yield item[0]
 | 
						|
 | 
						|
 | 
						|
def get_opnames(ex):
 | 
						|
    return set(iter_opnames(ex))
 | 
						|
 | 
						|
 | 
						|
@requires_specialization
 | 
						|
class TestExecutorInvalidation(unittest.TestCase):
 | 
						|
 | 
						|
    def setUp(self):
 | 
						|
        self.old = _testinternalcapi.get_optimizer()
 | 
						|
        self.opt = _testinternalcapi.new_counter_optimizer()
 | 
						|
        _testinternalcapi.set_optimizer(self.opt)
 | 
						|
 | 
						|
    def tearDown(self):
 | 
						|
        _testinternalcapi.set_optimizer(self.old)
 | 
						|
 | 
						|
    def test_invalidate_object(self):
 | 
						|
        # Generate a new set of functions at each call
 | 
						|
        ns = {}
 | 
						|
        func_src = "\n".join(
 | 
						|
            f"""
 | 
						|
            def f{n}():
 | 
						|
                for _ in range(1000):
 | 
						|
                    pass
 | 
						|
            """ for n in range(5)
 | 
						|
        )
 | 
						|
        exec(textwrap.dedent(func_src), ns, ns)
 | 
						|
        funcs = [ ns[f'f{n}'] for n in range(5)]
 | 
						|
        objects = [object() for _ in range(5)]
 | 
						|
 | 
						|
        for f in funcs:
 | 
						|
            f()
 | 
						|
        executors = [get_first_executor(f) for f in funcs]
 | 
						|
        # Set things up so each executor depends on the objects
 | 
						|
        # with an equal or lower index.
 | 
						|
        for i, exe in enumerate(executors):
 | 
						|
            self.assertTrue(exe.is_valid())
 | 
						|
            for obj in objects[:i+1]:
 | 
						|
                _testinternalcapi.add_executor_dependency(exe, obj)
 | 
						|
            self.assertTrue(exe.is_valid())
 | 
						|
        # Assert that the correct executors are invalidated
 | 
						|
        # and check that nothing crashes when we invalidate
 | 
						|
        # an executor mutliple times.
 | 
						|
        for i in (4,3,2,1,0):
 | 
						|
            _testinternalcapi.invalidate_executors(objects[i])
 | 
						|
            for exe in executors[i:]:
 | 
						|
                self.assertFalse(exe.is_valid())
 | 
						|
            for exe in executors[:i]:
 | 
						|
                self.assertTrue(exe.is_valid())
 | 
						|
 | 
						|
    def test_uop_optimizer_invalidation(self):
 | 
						|
        # Generate a new function at each call
 | 
						|
        ns = {}
 | 
						|
        exec(textwrap.dedent("""
 | 
						|
            def f():
 | 
						|
                for i in range(1000):
 | 
						|
                    pass
 | 
						|
        """), ns, ns)
 | 
						|
        f = ns['f']
 | 
						|
        opt = _testinternalcapi.new_uop_optimizer()
 | 
						|
        with temporary_optimizer(opt):
 | 
						|
            f()
 | 
						|
        exe = get_first_executor(f)
 | 
						|
        self.assertIsNotNone(exe)
 | 
						|
        self.assertTrue(exe.is_valid())
 | 
						|
        _testinternalcapi.invalidate_executors(f.__code__)
 | 
						|
        self.assertFalse(exe.is_valid())
 | 
						|
 | 
						|
    def test_sys__clear_internal_caches(self):
 | 
						|
        def f():
 | 
						|
            for _ in range(1000):
 | 
						|
                pass
 | 
						|
        opt = _testinternalcapi.new_uop_optimizer()
 | 
						|
        with temporary_optimizer(opt):
 | 
						|
            f()
 | 
						|
        exe = get_first_executor(f)
 | 
						|
        self.assertIsNotNone(exe)
 | 
						|
        self.assertTrue(exe.is_valid())
 | 
						|
        sys._clear_internal_caches()
 | 
						|
        self.assertFalse(exe.is_valid())
 | 
						|
        exe = get_first_executor(f)
 | 
						|
        self.assertIsNone(exe)
 | 
						|
 | 
						|
 | 
						|
@requires_specialization
 | 
						|
@unittest.skipIf(os.getenv("PYTHON_UOPS_OPTIMIZE") == "0", "Needs uop optimizer to run.")
 | 
						|
class TestUops(unittest.TestCase):
 | 
						|
 | 
						|
    def test_basic_loop(self):
 | 
						|
        def testfunc(x):
 | 
						|
            i = 0
 | 
						|
            while i < x:
 | 
						|
                i += 1
 | 
						|
 | 
						|
        opt = _testinternalcapi.new_uop_optimizer()
 | 
						|
        with temporary_optimizer(opt):
 | 
						|
            testfunc(1000)
 | 
						|
 | 
						|
        ex = get_first_executor(testfunc)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        uops = get_opnames(ex)
 | 
						|
        self.assertIn("_SET_IP", uops)
 | 
						|
        self.assertIn("_LOAD_FAST_0", uops)
 | 
						|
 | 
						|
    def test_extended_arg(self):
 | 
						|
        "Check EXTENDED_ARG handling in superblock creation"
 | 
						|
        ns = {}
 | 
						|
        exec(textwrap.dedent("""
 | 
						|
            def many_vars():
 | 
						|
                # 260 vars, so z9 should have index 259
 | 
						|
                a0 = a1 = a2 = a3 = a4 = a5 = a6 = a7 = a8 = a9 = 42
 | 
						|
                b0 = b1 = b2 = b3 = b4 = b5 = b6 = b7 = b8 = b9 = 42
 | 
						|
                c0 = c1 = c2 = c3 = c4 = c5 = c6 = c7 = c8 = c9 = 42
 | 
						|
                d0 = d1 = d2 = d3 = d4 = d5 = d6 = d7 = d8 = d9 = 42
 | 
						|
                e0 = e1 = e2 = e3 = e4 = e5 = e6 = e7 = e8 = e9 = 42
 | 
						|
                f0 = f1 = f2 = f3 = f4 = f5 = f6 = f7 = f8 = f9 = 42
 | 
						|
                g0 = g1 = g2 = g3 = g4 = g5 = g6 = g7 = g8 = g9 = 42
 | 
						|
                h0 = h1 = h2 = h3 = h4 = h5 = h6 = h7 = h8 = h9 = 42
 | 
						|
                i0 = i1 = i2 = i3 = i4 = i5 = i6 = i7 = i8 = i9 = 42
 | 
						|
                j0 = j1 = j2 = j3 = j4 = j5 = j6 = j7 = j8 = j9 = 42
 | 
						|
                k0 = k1 = k2 = k3 = k4 = k5 = k6 = k7 = k8 = k9 = 42
 | 
						|
                l0 = l1 = l2 = l3 = l4 = l5 = l6 = l7 = l8 = l9 = 42
 | 
						|
                m0 = m1 = m2 = m3 = m4 = m5 = m6 = m7 = m8 = m9 = 42
 | 
						|
                n0 = n1 = n2 = n3 = n4 = n5 = n6 = n7 = n8 = n9 = 42
 | 
						|
                o0 = o1 = o2 = o3 = o4 = o5 = o6 = o7 = o8 = o9 = 42
 | 
						|
                p0 = p1 = p2 = p3 = p4 = p5 = p6 = p7 = p8 = p9 = 42
 | 
						|
                q0 = q1 = q2 = q3 = q4 = q5 = q6 = q7 = q8 = q9 = 42
 | 
						|
                r0 = r1 = r2 = r3 = r4 = r5 = r6 = r7 = r8 = r9 = 42
 | 
						|
                s0 = s1 = s2 = s3 = s4 = s5 = s6 = s7 = s8 = s9 = 42
 | 
						|
                t0 = t1 = t2 = t3 = t4 = t5 = t6 = t7 = t8 = t9 = 42
 | 
						|
                u0 = u1 = u2 = u3 = u4 = u5 = u6 = u7 = u8 = u9 = 42
 | 
						|
                v0 = v1 = v2 = v3 = v4 = v5 = v6 = v7 = v8 = v9 = 42
 | 
						|
                w0 = w1 = w2 = w3 = w4 = w5 = w6 = w7 = w8 = w9 = 42
 | 
						|
                x0 = x1 = x2 = x3 = x4 = x5 = x6 = x7 = x8 = x9 = 42
 | 
						|
                y0 = y1 = y2 = y3 = y4 = y5 = y6 = y7 = y8 = y9 = 42
 | 
						|
                z0 = z1 = z2 = z3 = z4 = z5 = z6 = z7 = z8 = z9 = 42
 | 
						|
                while z9 > 0:
 | 
						|
                    z9 = z9 - 1
 | 
						|
        """), ns, ns)
 | 
						|
        many_vars = ns["many_vars"]
 | 
						|
 | 
						|
        opt = _testinternalcapi.new_uop_optimizer()
 | 
						|
        with temporary_optimizer(opt):
 | 
						|
            ex = get_first_executor(many_vars)
 | 
						|
            self.assertIsNone(ex)
 | 
						|
            many_vars()
 | 
						|
 | 
						|
        ex = get_first_executor(many_vars)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        self.assertTrue(any((opcode, oparg, operand) == ("_LOAD_FAST", 259, 0)
 | 
						|
                            for opcode, oparg, _, operand in list(ex)))
 | 
						|
 | 
						|
    def test_unspecialized_unpack(self):
 | 
						|
        # An example of an unspecialized opcode
 | 
						|
        def testfunc(x):
 | 
						|
            i = 0
 | 
						|
            while i < x:
 | 
						|
                i += 1
 | 
						|
                a, b = {1: 2, 3: 3}
 | 
						|
            assert a == 1 and b == 3
 | 
						|
            i = 0
 | 
						|
            while i < x:
 | 
						|
                i += 1
 | 
						|
 | 
						|
        opt = _testinternalcapi.new_uop_optimizer()
 | 
						|
 | 
						|
        with temporary_optimizer(opt):
 | 
						|
            testfunc(20)
 | 
						|
 | 
						|
        ex = get_first_executor(testfunc)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        uops = get_opnames(ex)
 | 
						|
        self.assertIn("_UNPACK_SEQUENCE", uops)
 | 
						|
 | 
						|
    def test_pop_jump_if_false(self):
 | 
						|
        def testfunc(n):
 | 
						|
            i = 0
 | 
						|
            while i < n:
 | 
						|
                i += 1
 | 
						|
 | 
						|
        opt = _testinternalcapi.new_uop_optimizer()
 | 
						|
        with temporary_optimizer(opt):
 | 
						|
            testfunc(20)
 | 
						|
 | 
						|
        ex = get_first_executor(testfunc)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        uops = get_opnames(ex)
 | 
						|
        self.assertIn("_GUARD_IS_TRUE_POP", uops)
 | 
						|
 | 
						|
    def test_pop_jump_if_none(self):
 | 
						|
        def testfunc(a):
 | 
						|
            for x in a:
 | 
						|
                if x is None:
 | 
						|
                    x = 0
 | 
						|
 | 
						|
        opt = _testinternalcapi.new_uop_optimizer()
 | 
						|
        with temporary_optimizer(opt):
 | 
						|
            testfunc(range(20))
 | 
						|
 | 
						|
        ex = get_first_executor(testfunc)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        uops = get_opnames(ex)
 | 
						|
        self.assertNotIn("_GUARD_IS_NONE_POP", uops)
 | 
						|
        self.assertNotIn("_GUARD_IS_NOT_NONE_POP", uops)
 | 
						|
 | 
						|
    def test_pop_jump_if_not_none(self):
 | 
						|
        def testfunc(a):
 | 
						|
            for x in a:
 | 
						|
                x = None
 | 
						|
                if x is not None:
 | 
						|
                    x = 0
 | 
						|
 | 
						|
        opt = _testinternalcapi.new_uop_optimizer()
 | 
						|
        with temporary_optimizer(opt):
 | 
						|
            testfunc(range(20))
 | 
						|
 | 
						|
        ex = get_first_executor(testfunc)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        uops = get_opnames(ex)
 | 
						|
        self.assertNotIn("_GUARD_IS_NONE_POP", uops)
 | 
						|
        self.assertNotIn("_GUARD_IS_NOT_NONE_POP", uops)
 | 
						|
 | 
						|
    def test_pop_jump_if_true(self):
 | 
						|
        def testfunc(n):
 | 
						|
            i = 0
 | 
						|
            while not i >= n:
 | 
						|
                i += 1
 | 
						|
 | 
						|
        opt = _testinternalcapi.new_uop_optimizer()
 | 
						|
        with temporary_optimizer(opt):
 | 
						|
            testfunc(20)
 | 
						|
 | 
						|
        ex = get_first_executor(testfunc)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        uops = get_opnames(ex)
 | 
						|
        self.assertIn("_GUARD_IS_FALSE_POP", uops)
 | 
						|
 | 
						|
    def test_jump_backward(self):
 | 
						|
        def testfunc(n):
 | 
						|
            i = 0
 | 
						|
            while i < n:
 | 
						|
                i += 1
 | 
						|
 | 
						|
        opt = _testinternalcapi.new_uop_optimizer()
 | 
						|
        with temporary_optimizer(opt):
 | 
						|
            testfunc(20)
 | 
						|
 | 
						|
        ex = get_first_executor(testfunc)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        uops = get_opnames(ex)
 | 
						|
        self.assertIn("_JUMP_TO_TOP", uops)
 | 
						|
 | 
						|
    def test_jump_forward(self):
 | 
						|
        def testfunc(n):
 | 
						|
            a = 0
 | 
						|
            while a < n:
 | 
						|
                if a < 0:
 | 
						|
                    a = -a
 | 
						|
                else:
 | 
						|
                    a = +a
 | 
						|
                a += 1
 | 
						|
            return a
 | 
						|
 | 
						|
        opt = _testinternalcapi.new_uop_optimizer()
 | 
						|
        with temporary_optimizer(opt):
 | 
						|
            testfunc(20)
 | 
						|
 | 
						|
        ex = get_first_executor(testfunc)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        uops = get_opnames(ex)
 | 
						|
        # Since there is no JUMP_FORWARD instruction,
 | 
						|
        # look for indirect evidence: the += operator
 | 
						|
        self.assertIn("_BINARY_OP_ADD_INT", uops)
 | 
						|
 | 
						|
    def test_for_iter_range(self):
 | 
						|
        def testfunc(n):
 | 
						|
            total = 0
 | 
						|
            for i in range(n):
 | 
						|
                total += i
 | 
						|
            return total
 | 
						|
 | 
						|
        opt = _testinternalcapi.new_uop_optimizer()
 | 
						|
        with temporary_optimizer(opt):
 | 
						|
            total = testfunc(20)
 | 
						|
            self.assertEqual(total, 190)
 | 
						|
 | 
						|
        ex = get_first_executor(testfunc)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        # for i, (opname, oparg) in enumerate(ex):
 | 
						|
        #     print(f"{i:4d}: {opname:<20s} {oparg:3d}")
 | 
						|
        uops = get_opnames(ex)
 | 
						|
        self.assertIn("_GUARD_NOT_EXHAUSTED_RANGE", uops)
 | 
						|
        # Verification that the jump goes past END_FOR
 | 
						|
        # is done by manual inspection of the output
 | 
						|
 | 
						|
    def test_for_iter_list(self):
 | 
						|
        def testfunc(a):
 | 
						|
            total = 0
 | 
						|
            for i in a:
 | 
						|
                total += i
 | 
						|
            return total
 | 
						|
 | 
						|
        opt = _testinternalcapi.new_uop_optimizer()
 | 
						|
        with temporary_optimizer(opt):
 | 
						|
            a = list(range(20))
 | 
						|
            total = testfunc(a)
 | 
						|
            self.assertEqual(total, 190)
 | 
						|
 | 
						|
        ex = get_first_executor(testfunc)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        # for i, (opname, oparg) in enumerate(ex):
 | 
						|
        #     print(f"{i:4d}: {opname:<20s} {oparg:3d}")
 | 
						|
        uops = get_opnames(ex)
 | 
						|
        self.assertIn("_GUARD_NOT_EXHAUSTED_LIST", uops)
 | 
						|
        # Verification that the jump goes past END_FOR
 | 
						|
        # is done by manual inspection of the output
 | 
						|
 | 
						|
    def test_for_iter_tuple(self):
 | 
						|
        def testfunc(a):
 | 
						|
            total = 0
 | 
						|
            for i in a:
 | 
						|
                total += i
 | 
						|
            return total
 | 
						|
 | 
						|
        opt = _testinternalcapi.new_uop_optimizer()
 | 
						|
        with temporary_optimizer(opt):
 | 
						|
            a = tuple(range(20))
 | 
						|
            total = testfunc(a)
 | 
						|
            self.assertEqual(total, 190)
 | 
						|
 | 
						|
        ex = get_first_executor(testfunc)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        # for i, (opname, oparg) in enumerate(ex):
 | 
						|
        #     print(f"{i:4d}: {opname:<20s} {oparg:3d}")
 | 
						|
        uops = get_opnames(ex)
 | 
						|
        self.assertIn("_GUARD_NOT_EXHAUSTED_TUPLE", uops)
 | 
						|
        # Verification that the jump goes past END_FOR
 | 
						|
        # is done by manual inspection of the output
 | 
						|
 | 
						|
    def test_list_edge_case(self):
 | 
						|
        def testfunc(it):
 | 
						|
            for x in it:
 | 
						|
                pass
 | 
						|
 | 
						|
        opt = _testinternalcapi.new_uop_optimizer()
 | 
						|
        with temporary_optimizer(opt):
 | 
						|
            a = [1, 2, 3]
 | 
						|
            it = iter(a)
 | 
						|
            testfunc(it)
 | 
						|
            a.append(4)
 | 
						|
            with self.assertRaises(StopIteration):
 | 
						|
                next(it)
 | 
						|
 | 
						|
    def test_call_py_exact_args(self):
 | 
						|
        def testfunc(n):
 | 
						|
            def dummy(x):
 | 
						|
                return x+1
 | 
						|
            for i in range(n):
 | 
						|
                dummy(i)
 | 
						|
 | 
						|
        opt = _testinternalcapi.new_uop_optimizer()
 | 
						|
        with temporary_optimizer(opt):
 | 
						|
            testfunc(20)
 | 
						|
 | 
						|
        ex = get_first_executor(testfunc)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        uops = get_opnames(ex)
 | 
						|
        self.assertIn("_PUSH_FRAME", uops)
 | 
						|
        self.assertIn("_BINARY_OP_ADD_INT", uops)
 | 
						|
 | 
						|
    def test_branch_taken(self):
 | 
						|
        def testfunc(n):
 | 
						|
            for i in range(n):
 | 
						|
                if i < 0:
 | 
						|
                    i = 0
 | 
						|
                else:
 | 
						|
                    i = 1
 | 
						|
 | 
						|
        opt = _testinternalcapi.new_uop_optimizer()
 | 
						|
        with temporary_optimizer(opt):
 | 
						|
            testfunc(20)
 | 
						|
 | 
						|
        ex = get_first_executor(testfunc)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        uops = get_opnames(ex)
 | 
						|
        self.assertIn("_GUARD_IS_FALSE_POP", uops)
 | 
						|
 | 
						|
    def test_for_iter_tier_two(self):
 | 
						|
        class MyIter:
 | 
						|
            def __init__(self, n):
 | 
						|
                self.n = n
 | 
						|
            def __iter__(self):
 | 
						|
                return self
 | 
						|
            def __next__(self):
 | 
						|
                self.n -= 1
 | 
						|
                if self.n < 0:
 | 
						|
                    raise StopIteration
 | 
						|
                return self.n
 | 
						|
 | 
						|
        def testfunc(n, m):
 | 
						|
            x = 0
 | 
						|
            for i in range(m):
 | 
						|
                for j in MyIter(n):
 | 
						|
                    x += 1000*i + j
 | 
						|
            return x
 | 
						|
 | 
						|
        opt = _testinternalcapi.new_uop_optimizer()
 | 
						|
        with temporary_optimizer(opt):
 | 
						|
            x = testfunc(10, 10)
 | 
						|
 | 
						|
        self.assertEqual(x, sum(range(10)) * 10010)
 | 
						|
 | 
						|
        ex = get_first_executor(testfunc)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        uops = get_opnames(ex)
 | 
						|
        self.assertIn("_FOR_ITER_TIER_TWO", uops)
 | 
						|
 | 
						|
    def test_confidence_score(self):
 | 
						|
        def testfunc(n):
 | 
						|
            bits = 0
 | 
						|
            for i in range(n):
 | 
						|
                if i & 0x01:
 | 
						|
                    bits += 1
 | 
						|
                if i & 0x02:
 | 
						|
                    bits += 1
 | 
						|
                if i&0x04:
 | 
						|
                    bits += 1
 | 
						|
                if i&0x08:
 | 
						|
                    bits += 1
 | 
						|
                if i&0x10:
 | 
						|
                    bits += 1
 | 
						|
                if i&0x20:
 | 
						|
                    bits += 1
 | 
						|
            return bits
 | 
						|
 | 
						|
        opt = _testinternalcapi.new_uop_optimizer()
 | 
						|
        with temporary_optimizer(opt):
 | 
						|
            x = testfunc(20)
 | 
						|
 | 
						|
        self.assertEqual(x, 40)
 | 
						|
        ex = get_first_executor(testfunc)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        ops = list(iter_opnames(ex))
 | 
						|
        #Since branch is 50/50 the trace could go either way.
 | 
						|
        count = ops.count("_GUARD_IS_TRUE_POP") + ops.count("_GUARD_IS_FALSE_POP")
 | 
						|
        self.assertLessEqual(count, 2)
 | 
						|
 | 
						|
 | 
						|
@requires_specialization
 | 
						|
@unittest.skipIf(os.getenv("PYTHON_UOPS_OPTIMIZE") == "0", "Needs uop optimizer to run.")
 | 
						|
class TestUopsOptimization(unittest.TestCase):
 | 
						|
 | 
						|
    def _run_with_optimizer(self, testfunc, arg):
 | 
						|
        res = None
 | 
						|
        opt = _testinternalcapi.new_uop_optimizer()
 | 
						|
        with temporary_optimizer(opt):
 | 
						|
            res = testfunc(arg)
 | 
						|
 | 
						|
        ex = get_first_executor(testfunc)
 | 
						|
        return res, ex
 | 
						|
 | 
						|
 | 
						|
    def test_int_type_propagation(self):
 | 
						|
        def testfunc(loops):
 | 
						|
            num = 0
 | 
						|
            for i in range(loops):
 | 
						|
                x = num + num
 | 
						|
                a = x + 1
 | 
						|
                num += 1
 | 
						|
            return a
 | 
						|
 | 
						|
        res, ex = self._run_with_optimizer(testfunc, 32)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        self.assertEqual(res, 63)
 | 
						|
        binop_count = [opname for opname in iter_opnames(ex) if opname == "_BINARY_OP_ADD_INT"]
 | 
						|
        guard_both_int_count = [opname for opname in iter_opnames(ex) if opname == "_GUARD_BOTH_INT"]
 | 
						|
        self.assertGreaterEqual(len(binop_count), 3)
 | 
						|
        self.assertLessEqual(len(guard_both_int_count), 1)
 | 
						|
 | 
						|
    def test_int_type_propagation_through_frame(self):
 | 
						|
        def double(x):
 | 
						|
            return x + x
 | 
						|
        def testfunc(loops):
 | 
						|
            num = 0
 | 
						|
            for i in range(loops):
 | 
						|
                x = num + num
 | 
						|
                a = double(x)
 | 
						|
                num += 1
 | 
						|
            return a
 | 
						|
 | 
						|
        opt = _testinternalcapi.new_uop_optimizer()
 | 
						|
        res = None
 | 
						|
        with temporary_optimizer(opt):
 | 
						|
            res = testfunc(32)
 | 
						|
 | 
						|
        ex = get_first_executor(testfunc)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        self.assertEqual(res, 124)
 | 
						|
        binop_count = [opname for opname in iter_opnames(ex) if opname == "_BINARY_OP_ADD_INT"]
 | 
						|
        guard_both_int_count = [opname for opname in iter_opnames(ex) if opname == "_GUARD_BOTH_INT"]
 | 
						|
        self.assertGreaterEqual(len(binop_count), 3)
 | 
						|
        self.assertLessEqual(len(guard_both_int_count), 1)
 | 
						|
 | 
						|
    def test_int_type_propagation_from_frame(self):
 | 
						|
        def double(x):
 | 
						|
            return x + x
 | 
						|
        def testfunc(loops):
 | 
						|
            num = 0
 | 
						|
            for i in range(loops):
 | 
						|
                a = double(num)
 | 
						|
                x = a + a
 | 
						|
                num += 1
 | 
						|
            return x
 | 
						|
 | 
						|
        opt = _testinternalcapi.new_uop_optimizer()
 | 
						|
        res = None
 | 
						|
        with temporary_optimizer(opt):
 | 
						|
            res = testfunc(32)
 | 
						|
 | 
						|
        ex = get_first_executor(testfunc)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        self.assertEqual(res, 124)
 | 
						|
        binop_count = [opname for opname in iter_opnames(ex) if opname == "_BINARY_OP_ADD_INT"]
 | 
						|
        guard_both_int_count = [opname for opname in iter_opnames(ex) if opname == "_GUARD_BOTH_INT"]
 | 
						|
        self.assertGreaterEqual(len(binop_count), 3)
 | 
						|
        self.assertLessEqual(len(guard_both_int_count), 1)
 | 
						|
 | 
						|
    def test_int_impure_region(self):
 | 
						|
        def testfunc(loops):
 | 
						|
            num = 0
 | 
						|
            while num < loops:
 | 
						|
                x = num + num
 | 
						|
                y = 1
 | 
						|
                x // 2
 | 
						|
                a = x + y
 | 
						|
                num += 1
 | 
						|
            return a
 | 
						|
 | 
						|
        res, ex = self._run_with_optimizer(testfunc, 64)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        binop_count = [opname for opname in iter_opnames(ex) if opname == "_BINARY_OP_ADD_INT"]
 | 
						|
        self.assertGreaterEqual(len(binop_count), 3)
 | 
						|
 | 
						|
    def test_call_py_exact_args(self):
 | 
						|
        def testfunc(n):
 | 
						|
            def dummy(x):
 | 
						|
                return x+1
 | 
						|
            for i in range(n):
 | 
						|
                dummy(i)
 | 
						|
 | 
						|
        res, ex = self._run_with_optimizer(testfunc, 32)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        uops = get_opnames(ex)
 | 
						|
        self.assertIn("_PUSH_FRAME", uops)
 | 
						|
        self.assertIn("_BINARY_OP_ADD_INT", uops)
 | 
						|
        self.assertNotIn("_CHECK_PEP_523", uops)
 | 
						|
 | 
						|
    def test_int_type_propagate_through_range(self):
 | 
						|
        def testfunc(n):
 | 
						|
 | 
						|
            for i in range(n):
 | 
						|
                x = i + i
 | 
						|
            return x
 | 
						|
 | 
						|
        res, ex = self._run_with_optimizer(testfunc, 32)
 | 
						|
        self.assertEqual(res, 62)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        uops = get_opnames(ex)
 | 
						|
        self.assertNotIn("_GUARD_BOTH_INT", uops)
 | 
						|
 | 
						|
    def test_int_value_numbering(self):
 | 
						|
        def testfunc(n):
 | 
						|
 | 
						|
            y = 1
 | 
						|
            for i in range(n):
 | 
						|
                x = y
 | 
						|
                z = x
 | 
						|
                a = z
 | 
						|
                b = a
 | 
						|
                res = x + z + a + b
 | 
						|
            return res
 | 
						|
 | 
						|
        res, ex = self._run_with_optimizer(testfunc, 32)
 | 
						|
        self.assertEqual(res, 4)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        uops = get_opnames(ex)
 | 
						|
        self.assertIn("_GUARD_BOTH_INT", uops)
 | 
						|
        guard_count = [opname for opname in iter_opnames(ex) if opname == "_GUARD_BOTH_INT"]
 | 
						|
        self.assertEqual(len(guard_count), 1)
 | 
						|
 | 
						|
    def test_comprehension(self):
 | 
						|
        def testfunc(n):
 | 
						|
            for _ in range(n):
 | 
						|
                return [i for i in range(n)]
 | 
						|
 | 
						|
        res, ex = self._run_with_optimizer(testfunc, 32)
 | 
						|
        self.assertEqual(res, list(range(32)))
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        uops = get_opnames(ex)
 | 
						|
        self.assertNotIn("_BINARY_OP_ADD_INT", uops)
 | 
						|
 | 
						|
    def test_call_py_exact_args_disappearing(self):
 | 
						|
        def dummy(x):
 | 
						|
            return x+1
 | 
						|
 | 
						|
        def testfunc(n):
 | 
						|
            for i in range(n):
 | 
						|
                dummy(i)
 | 
						|
 | 
						|
        opt = _testinternalcapi.new_uop_optimizer()
 | 
						|
        # Trigger specialization
 | 
						|
        testfunc(8)
 | 
						|
        with temporary_optimizer(opt):
 | 
						|
            del dummy
 | 
						|
            gc.collect()
 | 
						|
 | 
						|
            def dummy(x):
 | 
						|
                return x + 2
 | 
						|
            testfunc(32)
 | 
						|
 | 
						|
        ex = get_first_executor(testfunc)
 | 
						|
        # Honestly as long as it doesn't crash it's fine.
 | 
						|
        # Whether we get an executor or not is non-deterministic,
 | 
						|
        # because it's decided by when the function is freed.
 | 
						|
        # This test is a little implementation specific.
 | 
						|
 | 
						|
    def test_promote_globals_to_constants(self):
 | 
						|
 | 
						|
        result = script_helper.run_python_until_end('-c', textwrap.dedent("""
 | 
						|
        import _testinternalcapi
 | 
						|
        import opcode
 | 
						|
        import _opcode
 | 
						|
 | 
						|
        def get_first_executor(func):
 | 
						|
            code = func.__code__
 | 
						|
            co_code = code.co_code
 | 
						|
            for i in range(0, len(co_code), 2):
 | 
						|
                try:
 | 
						|
                    return _opcode.get_executor(code, i)
 | 
						|
                except ValueError:
 | 
						|
                    pass
 | 
						|
            return None
 | 
						|
 | 
						|
        def get_opnames(ex):
 | 
						|
            return {item[0] for item in ex}
 | 
						|
 | 
						|
        def testfunc(n):
 | 
						|
            for i in range(n):
 | 
						|
                x = range(i)
 | 
						|
            return x
 | 
						|
 | 
						|
        opt = _testinternalcapi.new_uop_optimizer()
 | 
						|
        _testinternalcapi.set_optimizer(opt)
 | 
						|
        testfunc(64)
 | 
						|
 | 
						|
        ex = get_first_executor(testfunc)
 | 
						|
        assert ex is not None
 | 
						|
        uops = get_opnames(ex)
 | 
						|
        assert "_LOAD_GLOBAL_BUILTINS" not in uops
 | 
						|
        assert "_LOAD_CONST_INLINE_BORROW_WITH_NULL" in uops
 | 
						|
        """))
 | 
						|
        self.assertEqual(result[0].rc, 0, result)
 | 
						|
 | 
						|
    def test_float_add_constant_propagation(self):
 | 
						|
        def testfunc(n):
 | 
						|
            a = 1.0
 | 
						|
            for _ in range(n):
 | 
						|
                a = a + 0.25
 | 
						|
                a = a + 0.25
 | 
						|
                a = a + 0.25
 | 
						|
                a = a + 0.25
 | 
						|
            return a
 | 
						|
 | 
						|
        res, ex = self._run_with_optimizer(testfunc, 32)
 | 
						|
        self.assertAlmostEqual(res, 33.0)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        uops = get_opnames(ex)
 | 
						|
        guard_both_float_count = [opname for opname in iter_opnames(ex) if opname == "_GUARD_BOTH_FLOAT"]
 | 
						|
        self.assertLessEqual(len(guard_both_float_count), 1)
 | 
						|
        # TODO gh-115506: this assertion may change after propagating constants.
 | 
						|
        # We'll also need to verify that propagation actually occurs.
 | 
						|
        self.assertIn("_BINARY_OP_ADD_FLOAT", uops)
 | 
						|
 | 
						|
    def test_float_subtract_constant_propagation(self):
 | 
						|
        def testfunc(n):
 | 
						|
            a = 1.0
 | 
						|
            for _ in range(n):
 | 
						|
                a = a - 0.25
 | 
						|
                a = a - 0.25
 | 
						|
                a = a - 0.25
 | 
						|
                a = a - 0.25
 | 
						|
            return a
 | 
						|
 | 
						|
        res, ex = self._run_with_optimizer(testfunc, 32)
 | 
						|
        self.assertAlmostEqual(res, -31.0)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        uops = get_opnames(ex)
 | 
						|
        guard_both_float_count = [opname for opname in iter_opnames(ex) if opname == "_GUARD_BOTH_FLOAT"]
 | 
						|
        self.assertLessEqual(len(guard_both_float_count), 1)
 | 
						|
        # TODO gh-115506: this assertion may change after propagating constants.
 | 
						|
        # We'll also need to verify that propagation actually occurs.
 | 
						|
        self.assertIn("_BINARY_OP_SUBTRACT_FLOAT", uops)
 | 
						|
 | 
						|
    def test_float_multiply_constant_propagation(self):
 | 
						|
        def testfunc(n):
 | 
						|
            a = 1.0
 | 
						|
            for _ in range(n):
 | 
						|
                a = a * 1.0
 | 
						|
                a = a * 1.0
 | 
						|
                a = a * 1.0
 | 
						|
                a = a * 1.0
 | 
						|
            return a
 | 
						|
 | 
						|
        res, ex = self._run_with_optimizer(testfunc, 32)
 | 
						|
        self.assertAlmostEqual(res, 1.0)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        uops = get_opnames(ex)
 | 
						|
        guard_both_float_count = [opname for opname in iter_opnames(ex) if opname == "_GUARD_BOTH_FLOAT"]
 | 
						|
        self.assertLessEqual(len(guard_both_float_count), 1)
 | 
						|
        # TODO gh-115506: this assertion may change after propagating constants.
 | 
						|
        # We'll also need to verify that propagation actually occurs.
 | 
						|
        self.assertIn("_BINARY_OP_MULTIPLY_FLOAT", uops)
 | 
						|
 | 
						|
    def test_add_unicode_propagation(self):
 | 
						|
        def testfunc(n):
 | 
						|
            a = ""
 | 
						|
            for _ in range(n):
 | 
						|
                a + a
 | 
						|
                a + a
 | 
						|
                a + a
 | 
						|
                a + a
 | 
						|
            return a
 | 
						|
 | 
						|
        res, ex = self._run_with_optimizer(testfunc, 32)
 | 
						|
        self.assertEqual(res, "")
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        uops = get_opnames(ex)
 | 
						|
        guard_both_unicode_count = [opname for opname in iter_opnames(ex) if opname == "_GUARD_BOTH_UNICODE"]
 | 
						|
        self.assertLessEqual(len(guard_both_unicode_count), 1)
 | 
						|
        self.assertIn("_BINARY_OP_ADD_UNICODE", uops)
 | 
						|
 | 
						|
    def test_compare_op_type_propagation_float(self):
 | 
						|
        def testfunc(n):
 | 
						|
            a = 1.0
 | 
						|
            for _ in range(n):
 | 
						|
                x = a == a
 | 
						|
                x = a == a
 | 
						|
                x = a == a
 | 
						|
                x = a == a
 | 
						|
            return x
 | 
						|
 | 
						|
        res, ex = self._run_with_optimizer(testfunc, 32)
 | 
						|
        self.assertTrue(res)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        uops = get_opnames(ex)
 | 
						|
        guard_both_float_count = [opname for opname in iter_opnames(ex) if opname == "_GUARD_BOTH_FLOAT"]
 | 
						|
        self.assertLessEqual(len(guard_both_float_count), 1)
 | 
						|
        self.assertIn("_COMPARE_OP_FLOAT", uops)
 | 
						|
 | 
						|
    def test_compare_op_type_propagation_int(self):
 | 
						|
        def testfunc(n):
 | 
						|
            a = 1
 | 
						|
            for _ in range(n):
 | 
						|
                x = a == a
 | 
						|
                x = a == a
 | 
						|
                x = a == a
 | 
						|
                x = a == a
 | 
						|
            return x
 | 
						|
 | 
						|
        res, ex = self._run_with_optimizer(testfunc, 32)
 | 
						|
        self.assertTrue(res)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        uops = get_opnames(ex)
 | 
						|
        guard_both_float_count = [opname for opname in iter_opnames(ex) if opname == "_GUARD_BOTH_INT"]
 | 
						|
        self.assertLessEqual(len(guard_both_float_count), 1)
 | 
						|
        self.assertIn("_COMPARE_OP_INT", uops)
 | 
						|
 | 
						|
    def test_compare_op_type_propagation_unicode(self):
 | 
						|
        def testfunc(n):
 | 
						|
            a = ""
 | 
						|
            for _ in range(n):
 | 
						|
                x = a == a
 | 
						|
                x = a == a
 | 
						|
                x = a == a
 | 
						|
                x = a == a
 | 
						|
            return x
 | 
						|
 | 
						|
        res, ex = self._run_with_optimizer(testfunc, 32)
 | 
						|
        self.assertTrue(res)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        uops = get_opnames(ex)
 | 
						|
        guard_both_float_count = [opname for opname in iter_opnames(ex) if opname == "_GUARD_BOTH_UNICODE"]
 | 
						|
        self.assertLessEqual(len(guard_both_float_count), 1)
 | 
						|
        self.assertIn("_COMPARE_OP_STR", uops)
 | 
						|
 | 
						|
    def test_type_inconsistency(self):
 | 
						|
        ns = {}
 | 
						|
        src = textwrap.dedent("""
 | 
						|
            def testfunc(n):
 | 
						|
                for i in range(n):
 | 
						|
                    x = _test_global + _test_global
 | 
						|
        """)
 | 
						|
        exec(src, ns, ns)
 | 
						|
        testfunc = ns['testfunc']
 | 
						|
        ns['_test_global'] = 0
 | 
						|
        _, ex = self._run_with_optimizer(testfunc, 16)
 | 
						|
        self.assertIsNone(ex)
 | 
						|
        ns['_test_global'] = 1
 | 
						|
        _, ex = self._run_with_optimizer(testfunc, 16)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
        uops = get_opnames(ex)
 | 
						|
        self.assertNotIn("_GUARD_BOTH_INT", uops)
 | 
						|
        self.assertIn("_BINARY_OP_ADD_INT", uops)
 | 
						|
        # Try again, but between the runs, set the global to a float.
 | 
						|
        # This should result in no executor the second time.
 | 
						|
        ns = {}
 | 
						|
        exec(src, ns, ns)
 | 
						|
        testfunc = ns['testfunc']
 | 
						|
        ns['_test_global'] = 0
 | 
						|
        _, ex = self._run_with_optimizer(testfunc, 16)
 | 
						|
        self.assertIsNone(ex)
 | 
						|
        ns['_test_global'] = 3.14
 | 
						|
        _, ex = self._run_with_optimizer(testfunc, 16)
 | 
						|
        self.assertIsNone(ex)
 | 
						|
 | 
						|
    def test_combine_stack_space_checks_sequential(self):
 | 
						|
        def dummy12(x):
 | 
						|
            return x - 1
 | 
						|
        def dummy13(y):
 | 
						|
            z = y + 2
 | 
						|
            return y, z
 | 
						|
        def testfunc(n):
 | 
						|
            a = 0
 | 
						|
            for _ in range(n):
 | 
						|
                b = dummy12(7)
 | 
						|
                c, d = dummy13(9)
 | 
						|
                a += b + c + d
 | 
						|
            return a
 | 
						|
 | 
						|
        res, ex = self._run_with_optimizer(testfunc, 32)
 | 
						|
        self.assertEqual(res, 832)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
 | 
						|
        uops_and_operands = [(opcode, operand) for opcode, _, _, operand in ex]
 | 
						|
        uop_names = [uop[0] for uop in uops_and_operands]
 | 
						|
        self.assertEqual(uop_names.count("_PUSH_FRAME"), 2)
 | 
						|
        self.assertEqual(uop_names.count("_POP_FRAME"), 2)
 | 
						|
        self.assertEqual(uop_names.count("_CHECK_STACK_SPACE"), 0)
 | 
						|
        self.assertEqual(uop_names.count("_CHECK_STACK_SPACE_OPERAND"), 1)
 | 
						|
        # sequential calls: max(12, 13) == 13
 | 
						|
        largest_stack = _testinternalcapi.get_co_framesize(dummy13.__code__)
 | 
						|
        self.assertIn(("_CHECK_STACK_SPACE_OPERAND", largest_stack), uops_and_operands)
 | 
						|
 | 
						|
    def test_combine_stack_space_checks_nested(self):
 | 
						|
        def dummy12(x):
 | 
						|
            return x + 3
 | 
						|
        def dummy15(y):
 | 
						|
            z = dummy12(y)
 | 
						|
            return y, z
 | 
						|
        def testfunc(n):
 | 
						|
            a = 0
 | 
						|
            for _ in range(n):
 | 
						|
                b, c = dummy15(2)
 | 
						|
                a += b + c
 | 
						|
            return a
 | 
						|
 | 
						|
        res, ex = self._run_with_optimizer(testfunc, 32)
 | 
						|
        self.assertEqual(res, 224)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
 | 
						|
        uops_and_operands = [(opcode, operand) for opcode, _, _, operand in ex]
 | 
						|
        uop_names = [uop[0] for uop in uops_and_operands]
 | 
						|
        self.assertEqual(uop_names.count("_PUSH_FRAME"), 2)
 | 
						|
        self.assertEqual(uop_names.count("_POP_FRAME"), 2)
 | 
						|
        self.assertEqual(uop_names.count("_CHECK_STACK_SPACE"), 0)
 | 
						|
        self.assertEqual(uop_names.count("_CHECK_STACK_SPACE_OPERAND"), 1)
 | 
						|
        # nested calls: 15 + 12 == 27
 | 
						|
        largest_stack = (
 | 
						|
            _testinternalcapi.get_co_framesize(dummy15.__code__) +
 | 
						|
            _testinternalcapi.get_co_framesize(dummy12.__code__)
 | 
						|
        )
 | 
						|
        self.assertIn(("_CHECK_STACK_SPACE_OPERAND", largest_stack), uops_and_operands)
 | 
						|
 | 
						|
    def test_combine_stack_space_checks_several_calls(self):
 | 
						|
        def dummy12(x):
 | 
						|
            return x + 3
 | 
						|
        def dummy13(y):
 | 
						|
            z = y + 2
 | 
						|
            return y, z
 | 
						|
        def dummy18(y):
 | 
						|
            z = dummy12(y)
 | 
						|
            x, w = dummy13(z)
 | 
						|
            return z, x, w
 | 
						|
        def testfunc(n):
 | 
						|
            a = 0
 | 
						|
            for _ in range(n):
 | 
						|
                b = dummy12(5)
 | 
						|
                c, d, e = dummy18(2)
 | 
						|
                a += b + c + d + e
 | 
						|
            return a
 | 
						|
 | 
						|
        res, ex = self._run_with_optimizer(testfunc, 32)
 | 
						|
        self.assertEqual(res, 800)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
 | 
						|
        uops_and_operands = [(opcode, operand) for opcode, _, _, operand in ex]
 | 
						|
        uop_names = [uop[0] for uop in uops_and_operands]
 | 
						|
        self.assertEqual(uop_names.count("_PUSH_FRAME"), 4)
 | 
						|
        self.assertEqual(uop_names.count("_POP_FRAME"), 4)
 | 
						|
        self.assertEqual(uop_names.count("_CHECK_STACK_SPACE"), 0)
 | 
						|
        self.assertEqual(uop_names.count("_CHECK_STACK_SPACE_OPERAND"), 1)
 | 
						|
        # max(12, 18 + max(12, 13)) == 31
 | 
						|
        largest_stack = (
 | 
						|
            _testinternalcapi.get_co_framesize(dummy18.__code__) +
 | 
						|
            _testinternalcapi.get_co_framesize(dummy13.__code__)
 | 
						|
        )
 | 
						|
        self.assertIn(("_CHECK_STACK_SPACE_OPERAND", largest_stack), uops_and_operands)
 | 
						|
 | 
						|
    def test_combine_stack_space_checks_several_calls_different_order(self):
 | 
						|
        # same as `several_calls` but with top-level calls reversed
 | 
						|
        def dummy12(x):
 | 
						|
            return x + 3
 | 
						|
        def dummy13(y):
 | 
						|
            z = y + 2
 | 
						|
            return y, z
 | 
						|
        def dummy18(y):
 | 
						|
            z = dummy12(y)
 | 
						|
            x, w = dummy13(z)
 | 
						|
            return z, x, w
 | 
						|
        def testfunc(n):
 | 
						|
            a = 0
 | 
						|
            for _ in range(n):
 | 
						|
                c, d, e = dummy18(2)
 | 
						|
                b = dummy12(5)
 | 
						|
                a += b + c + d + e
 | 
						|
            return a
 | 
						|
 | 
						|
        res, ex = self._run_with_optimizer(testfunc, 32)
 | 
						|
        self.assertEqual(res, 800)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
 | 
						|
        uops_and_operands = [(opcode, operand) for opcode, _, _, operand in ex]
 | 
						|
        uop_names = [uop[0] for uop in uops_and_operands]
 | 
						|
        self.assertEqual(uop_names.count("_PUSH_FRAME"), 4)
 | 
						|
        self.assertEqual(uop_names.count("_POP_FRAME"), 4)
 | 
						|
        self.assertEqual(uop_names.count("_CHECK_STACK_SPACE"), 0)
 | 
						|
        self.assertEqual(uop_names.count("_CHECK_STACK_SPACE_OPERAND"), 1)
 | 
						|
        # max(18 + max(12, 13), 12) == 31
 | 
						|
        largest_stack = (
 | 
						|
            _testinternalcapi.get_co_framesize(dummy18.__code__) +
 | 
						|
            _testinternalcapi.get_co_framesize(dummy13.__code__)
 | 
						|
        )
 | 
						|
        self.assertIn(("_CHECK_STACK_SPACE_OPERAND", largest_stack), uops_and_operands)
 | 
						|
 | 
						|
    def test_combine_stack_space_complex(self):
 | 
						|
        def dummy0(x):
 | 
						|
            return x
 | 
						|
        def dummy1(x):
 | 
						|
            return dummy0(x)
 | 
						|
        def dummy2(x):
 | 
						|
            return dummy1(x)
 | 
						|
        def dummy3(x):
 | 
						|
            return dummy0(x)
 | 
						|
        def dummy4(x):
 | 
						|
            y = dummy0(x)
 | 
						|
            return dummy3(y)
 | 
						|
        def dummy5(x):
 | 
						|
            return dummy2(x)
 | 
						|
        def dummy6(x):
 | 
						|
            y = dummy5(x)
 | 
						|
            z = dummy0(y)
 | 
						|
            return dummy4(z)
 | 
						|
        def testfunc(n):
 | 
						|
            a = 0;
 | 
						|
            for _ in range(32):
 | 
						|
                b = dummy5(1)
 | 
						|
                c = dummy0(1)
 | 
						|
                d = dummy6(1)
 | 
						|
                a += b + c + d
 | 
						|
            return a
 | 
						|
 | 
						|
        res, ex = self._run_with_optimizer(testfunc, 32)
 | 
						|
        self.assertEqual(res, 96)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
 | 
						|
        uops_and_operands = [(opcode, operand) for opcode, _, _, operand in ex]
 | 
						|
        uop_names = [uop[0] for uop in uops_and_operands]
 | 
						|
        self.assertEqual(uop_names.count("_PUSH_FRAME"), 15)
 | 
						|
        self.assertEqual(uop_names.count("_POP_FRAME"), 15)
 | 
						|
 | 
						|
        self.assertEqual(uop_names.count("_CHECK_STACK_SPACE"), 0)
 | 
						|
        self.assertEqual(uop_names.count("_CHECK_STACK_SPACE_OPERAND"), 1)
 | 
						|
        largest_stack = (
 | 
						|
            _testinternalcapi.get_co_framesize(dummy6.__code__) +
 | 
						|
            _testinternalcapi.get_co_framesize(dummy5.__code__) +
 | 
						|
            _testinternalcapi.get_co_framesize(dummy2.__code__) +
 | 
						|
            _testinternalcapi.get_co_framesize(dummy1.__code__) +
 | 
						|
            _testinternalcapi.get_co_framesize(dummy0.__code__)
 | 
						|
        )
 | 
						|
        self.assertIn(
 | 
						|
            ("_CHECK_STACK_SPACE_OPERAND", largest_stack), uops_and_operands
 | 
						|
        )
 | 
						|
 | 
						|
    def test_combine_stack_space_checks_large_framesize(self):
 | 
						|
        # Create a function with a large framesize. This ensures _CHECK_STACK_SPACE is
 | 
						|
        # actually doing its job. Note that the resulting trace hits
 | 
						|
        # UOP_MAX_TRACE_LENGTH, but since all _CHECK_STACK_SPACEs happen early, this
 | 
						|
        # test is still meaningful.
 | 
						|
        repetitions = 10000
 | 
						|
        ns = {}
 | 
						|
        header = """
 | 
						|
            def dummy_large(a0):
 | 
						|
        """
 | 
						|
        body = "".join([f"""
 | 
						|
                a{n+1} = a{n} + 1
 | 
						|
        """ for n in range(repetitions)])
 | 
						|
        return_ = f"""
 | 
						|
                return a{repetitions-1}
 | 
						|
        """
 | 
						|
        exec(textwrap.dedent(header + body + return_), ns, ns)
 | 
						|
        dummy_large = ns['dummy_large']
 | 
						|
 | 
						|
        # this is something like:
 | 
						|
        #
 | 
						|
        # def dummy_large(a0):
 | 
						|
        #     a1 = a0 + 1
 | 
						|
        #     a2 = a1 + 1
 | 
						|
        #     ....
 | 
						|
        #     a9999 = a9998 + 1
 | 
						|
        #     return a9999
 | 
						|
 | 
						|
        def dummy15(z):
 | 
						|
            y = dummy_large(z)
 | 
						|
            return y + 3
 | 
						|
 | 
						|
        def testfunc(n):
 | 
						|
            b = 0
 | 
						|
            for _ in range(n):
 | 
						|
                b += dummy15(7)
 | 
						|
            return b
 | 
						|
 | 
						|
        res, ex = self._run_with_optimizer(testfunc, 32)
 | 
						|
        self.assertEqual(res, 32 * (repetitions + 9))
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
 | 
						|
        uops_and_operands = [(opcode, operand) for opcode, _, _, operand in ex]
 | 
						|
        uop_names = [uop[0] for uop in uops_and_operands]
 | 
						|
        self.assertEqual(uop_names.count("_PUSH_FRAME"), 2)
 | 
						|
        self.assertEqual(uop_names.count("_CHECK_STACK_SPACE_OPERAND"), 1)
 | 
						|
 | 
						|
        # this hits a different case during trace projection in refcount test runs only,
 | 
						|
        # so we need to account for both possibilities
 | 
						|
        self.assertIn(uop_names.count("_CHECK_STACK_SPACE"), [0, 1])
 | 
						|
        if uop_names.count("_CHECK_STACK_SPACE") == 0:
 | 
						|
            largest_stack = (
 | 
						|
                _testinternalcapi.get_co_framesize(dummy15.__code__) +
 | 
						|
                _testinternalcapi.get_co_framesize(dummy_large.__code__)
 | 
						|
            )
 | 
						|
        else:
 | 
						|
            largest_stack = _testinternalcapi.get_co_framesize(dummy15.__code__)
 | 
						|
        self.assertIn(
 | 
						|
            ("_CHECK_STACK_SPACE_OPERAND", largest_stack), uops_and_operands
 | 
						|
        )
 | 
						|
 | 
						|
    def test_combine_stack_space_checks_recursion(self):
 | 
						|
        def dummy15(x):
 | 
						|
            while x > 0:
 | 
						|
                return dummy15(x - 1)
 | 
						|
            return 42
 | 
						|
        def testfunc(n):
 | 
						|
            a = 0
 | 
						|
            for _ in range(n):
 | 
						|
                a += dummy15(n)
 | 
						|
            return a
 | 
						|
 | 
						|
        res, ex = self._run_with_optimizer(testfunc, 32)
 | 
						|
        self.assertEqual(res, 42 * 32)
 | 
						|
        self.assertIsNotNone(ex)
 | 
						|
 | 
						|
        uops_and_operands = [(opcode, operand) for opcode, _, _, operand in ex]
 | 
						|
        uop_names = [uop[0] for uop in uops_and_operands]
 | 
						|
        self.assertEqual(uop_names.count("_PUSH_FRAME"), 2)
 | 
						|
        self.assertEqual(uop_names.count("_POP_FRAME"), 0)
 | 
						|
        self.assertEqual(uop_names.count("_CHECK_STACK_SPACE"), 1)
 | 
						|
        self.assertEqual(uop_names.count("_CHECK_STACK_SPACE_OPERAND"), 1)
 | 
						|
        largest_stack = _testinternalcapi.get_co_framesize(dummy15.__code__)
 | 
						|
        self.assertIn(("_CHECK_STACK_SPACE_OPERAND", largest_stack), uops_and_operands)
 | 
						|
 | 
						|
    def test_many_nested(self):
 | 
						|
        # overflow the trace_stack
 | 
						|
        def dummy_a(x):
 | 
						|
            return x
 | 
						|
        def dummy_b(x):
 | 
						|
            return dummy_a(x)
 | 
						|
        def dummy_c(x):
 | 
						|
            return dummy_b(x)
 | 
						|
        def dummy_d(x):
 | 
						|
            return dummy_c(x)
 | 
						|
        def dummy_e(x):
 | 
						|
            return dummy_d(x)
 | 
						|
        def dummy_f(x):
 | 
						|
            return dummy_e(x)
 | 
						|
        def dummy_g(x):
 | 
						|
            return dummy_f(x)
 | 
						|
        def dummy_h(x):
 | 
						|
            return dummy_g(x)
 | 
						|
        def testfunc(n):
 | 
						|
            a = 0
 | 
						|
            for _ in range(n):
 | 
						|
                a += dummy_h(n)
 | 
						|
            return a
 | 
						|
 | 
						|
        res, ex = self._run_with_optimizer(testfunc, 32)
 | 
						|
        self.assertEqual(res, 32 * 32)
 | 
						|
        self.assertIsNone(ex)
 | 
						|
 | 
						|
if __name__ == "__main__":
 | 
						|
    unittest.main()
 |