python · brandtbucher · Jan 29, 2025 · Jan 28, 2025
diff --git a/Lib/test/test_call.py b/Lib/test/test_call.py
@@ -1,6 +1,6 @@
 import unittest
 from test.support import (cpython_only, is_wasi, requires_limited_api, Py_DEBUG,
-                          set_recursion_limit, skip_on_s390x, skip_emscripten_stack_overflow)
+                          set_recursion_limit, skip_on_s390x, skip_emscripten_stack_overflow, import_helper)
 try:
     import _testcapi
 except ImportError:
@@ -616,9 +616,6 @@ def testfunction_kw(self, *, kw):
     return self
 
 
-ADAPTIVE_WARMUP_DELAY = 2
-
-
 @unittest.skipIf(_testcapi is None, "requires _testcapi")
 class TestPEP590(unittest.TestCase):
 
@@ -802,17 +799,18 @@ def __call__(self, *args):
 
     def test_setvectorcall(self):
         from _testcapi import function_setvectorcall
+        _testinternalcapi = import_helper.import_module("_testinternalcapi")
         def f(num): return num + 1
         assert_equal = self.assertEqual
         num = 10
         assert_equal(11, f(num))
         function_setvectorcall(f)
-        # make sure specializer is triggered by running > 50 times
-        for _ in range(10 * ADAPTIVE_WARMUP_DELAY):
+        for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD):
             assert_equal("overridden", f(num))
 
     def test_setvectorcall_load_attr_specialization_skip(self):
         from _testcapi import function_setvectorcall
+        _testinternalcapi = import_helper.import_module("_testinternalcapi")
 
         class X:
             def __getattribute__(self, attr):
@@ -824,11 +822,12 @@ def __getattribute__(self, attr):
         function_setvectorcall(X.__getattribute__)
         # make sure specialization doesn't trigger
         # when vectorcall is overridden
-        for _ in range(ADAPTIVE_WARMUP_DELAY):
+        for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD):
             assert_equal("overridden", x.a)
 
     def test_setvectorcall_load_attr_specialization_deopt(self):
         from _testcapi import function_setvectorcall
+        _testinternalcapi = import_helper.import_module("_testinternalcapi")
 
         class X:
             def __getattribute__(self, attr):
@@ -840,12 +839,12 @@ def get_a(x):
         assert_equal = self.assertEqual
         x = X()
         # trigger LOAD_ATTR_GETATTRIBUTE_OVERRIDDEN specialization
-        for _ in range(ADAPTIVE_WARMUP_DELAY):
+        for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD):
             assert_equal("a", get_a(x))
         function_setvectorcall(X.__getattribute__)
         # make sure specialized LOAD_ATTR_GETATTRIBUTE_OVERRIDDEN
         # gets deopted due to overridden vectorcall
-        for _ in range(ADAPTIVE_WARMUP_DELAY):
+        for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD):
             assert_equal("overridden", get_a(x))
 
     @requires_limited_api

diff --git a/Lib/test/test_dis.py b/Lib/test/test_dis.py
@@ -15,7 +15,7 @@
 import unittest
 from test.support import (captured_stdout, requires_debug_ranges,
                           requires_specialization, cpython_only,
-                          os_helper)
+                          os_helper, import_helper)
 from test.support.bytecode_helper import BytecodeTestCase
 
 
@@ -931,8 +931,6 @@ def extended_arg_quick():
 """% (extended_arg_quick.__code__.co_firstlineno,
       extended_arg_quick.__code__.co_firstlineno + 1,)
 
-ADAPTIVE_WARMUP_DELAY = 2
-
 class DisTestBase(unittest.TestCase):
     "Common utilities for DisTests and TestDisTraceback"
 
@@ -1259,8 +1257,9 @@ def test__try_compile_no_context_exc_on_error(self):
             self.assertIsNone(e.__context__)
 
     @staticmethod
-    def code_quicken(f, times=ADAPTIVE_WARMUP_DELAY):
-        for _ in range(times):
+    def code_quicken(f):
+        _testinternalcapi = import_helper.import_module("_testinternalcapi")
+        for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD):
             f()
 
     @cpython_only
@@ -1306,16 +1305,17 @@ def test_call_specialize(self):
     @requires_specialization
     def test_loop_quicken(self):
         # Loop can trigger a quicken where the loop is located
-        self.code_quicken(loop_test, 4)
+        self.code_quicken(loop_test)
         got = self.get_disassembly(loop_test, adaptive=True)
         expected = dis_loop_test_quickened_code
         self.do_disassembly_compare(got, expected)
 
     @cpython_only
     @requires_specialization
     def test_loop_with_conditional_at_end_is_quickened(self):
+        _testinternalcapi = import_helper.import_module("_testinternalcapi")
         def for_loop_true(x):
-       
8000
     for i in range(10):
+            for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD):
                 if x:
                     pass
 
@@ -1324,7 +1324,7 @@ def for_loop_true(x):
                       self.get_disassembly(for_loop_true, adaptive=True))
 
         def for_loop_false(x):
-            for i in range(10):
+            for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD):
                 if x:
                     pass
 
@@ -1334,7 +1334,7 @@ def for_loop_false(x):
 
         def while_loop():
             i = 0
-            while i < 10:
+            while i < _testinternalcapi.SPECIALIZATION_THRESHOLD:
                 i += 1
 
         while_loop()

diff --git a/Lib/test/test_embed.py b/Lib/test/test_embed.py
@@ -384,7 +384,9 @@ def test_simple_initialization_api(self):
     def test_specialized_static_code_gets_unspecialized_at_Py_FINALIZE(self):
         # https://github.com/python/cpython/issues/92031
 
-        code = textwrap.dedent("""\
+        _testinternalcapi = import_helper.import_module("_testinternalcapi")
+
+        code = textwrap.dedent(f"""\
             import dis
             import importlib._bootstrap
             import opcode
@@ -411,7 +413,7 @@ def is_specialized(f):
 
             assert not is_specialized(func), "specialized instructions found"
 
-            for i in range(test.test_dis.ADAPTIVE_WARMUP_DELAY):
+            for _ in range({_testinternalcapi.SPECIALIZATION_THRESHOLD}):
                 func(importlib._bootstrap, ["x"], lambda *args: None)
 
             assert is_specialized(func), "no specialized instructions found"

diff --git a/Lib/test/test_monitoring.py b/Lib/test/test_monitoring.py
@@ -14,6 +14,7 @@
 from test.support import requires_specialization_ft, script_helper
 
 _testcapi = test.support.import_helper.import_module("_testcapi")
+_testinternalcapi = test.support.import_helper.import_module("_testinternalcapi")
 
 PAIR = (0,1)
 
@@ -897,13 +898,13 @@ def implicit_stop_iteration(iterator=None):
         # re-specialize immediately, so that we can we can test the
         # unspecialized version of the loop first.
         # Note: this assumes that we don't specialize loops over sets.
-        implicit_stop_iteration(set(range(100)))
+        implicit_stop_iteration(set(range(_testinternalcapi.SPECIALIZATION_THRESHOLD)))
 
         # This will record a RAISE event for the StopIteration.
         self.check_events(implicit_stop_iteration, expected, recorders=recorders)
 
         # Now specialize, so that we see a STOP_ITERATION event.
-        for _ in range(100):
+        for _ in range(_testinternalcapi.SPECIALIZATION_COOLDOWN):
             implicit_stop_iteration()
 
         # This will record a STOP_ITERATION event for the StopIteration.
@@ -1057,7 +1058,7 @@ def f():
             except ValueError:
                 pass
 
-        for _ in range(100):
+        for _ in range(_testinternalcapi.SPECIALIZATION_THRESHOLD):
             f()
         recorders = (
             ReturnRecorder,
@@ -2033,8 +2034,8 @@ def __init__(self, set_event):
                     sys.monitoring.set_events(TEST_TOOL, E.PY_RESUME)
 
         def make_foo_optimized_then_set_event():
-            for i in range(100):
-                Foo(i == 99)
+            for i in range(_testinternalcapi.SPECIALIZATION_THRESHOLD + 1):
+                Foo(i == _testinternalcapi.SPECIALIZATION_THRESHOLD)
 
         try:
             make_foo_optimized_then_set_event()
@@ -2106,9 +2107,9 @@ def test_func(recorder):
             set_events = sys.monitoring.set_events
             line = E.LINE
             i = 0
-            for i in range(551):
-                # Turn on events without branching once i reaches 500.
-                set_events(TEST_TOOL, line * int(i >= 500))
+            for i in range(_testinternalcapi.SPECIALIZATION_THRESHOLD + 51):
+                # Turn on events without branching once i reaches _testinternalcapi.SPECIALIZATION_THRESHOLD.
+                set_events(TEST_TOOL, line * int(i >= _testinternalcapi.SPECIALIZATION_THRESHOLD))
                 pass
                 pass
                 pass