youknowone · Oct 28, 2026 · Oct 28, 2026 · Oct 28, 2026 · Oct 28, 2026
diff --git a/Lib/test/test_free_threading/__init__.py b/Lib/test/test_free_threading/__init__.py
@@ -0,0 +1,11 @@
+import os
+import unittest
+
+from test import support
+
+
+if not support.Py_GIL_DISABLED:
+    raise unittest.SkipTest("GIL enabled")
+
+def load_tests(*args):
+    return support.load_package_tests(os.path.dirname(__file__), *args)
diff --git a/Lib/test/test_free_threading/test_bisect.py b/Lib/test/test_free_threading/test_bisect.py
@@ -0,0 +1,56 @@
+import unittest
+from test.support import import_helper, threading_helper
+import random
+
+py_bisect = import_helper.import_fresh_module('bisect', blocked=['_bisect'])
+c_bisect = import_helper.import_fresh_module('bisect', fresh=['_bisect'])
+
+
+NTHREADS = 4
+OBJECT_COUNT = 500
+
+
+class TestBase:
+    def do_racing_insort(self, insert_method):
+        def insert(data):
+            for _ in range(OBJECT_COUNT):
+                x = random.randint(-OBJECT_COUNT, OBJECT_COUNT)
+                insert_method(data, x)
+
+        data = list(range(OBJECT_COUNT))
+        threading_helper.run_concurrently(
+            worker_func=insert, args=(data,), nthreads=NTHREADS
+        )
+        if False:
+            # These functions are not thread-safe and so the list can become
+            # unsorted.  However, we don't want Python to crash if these
+            # functions are used concurrently on the same sequence.  This
+            # should also not produce any TSAN warnings.
+            self.assertTrue(self.is_sorted_ascending(data))
+
+    def test_racing_insert_right(self):
+        self.do_racing_insort(self.mod.insort_right)
+
+    def test_racing_insert_left(self):
+        self.do_racing_insort(self.mod.insort_left)
+
+    @staticmethod
+    def is_sorted_ascending(lst):
+        """
+        Check if the list is sorted in ascending order (non-decreasing).
+        """
+        return all(lst[i - 1] <= lst[i] for i in range(1, len(lst)))
+
+
+@threading_helper.requires_working_threading()
+class TestPyBisect(unittest.TestCase, TestBase):
+    mod = py_bisect
+
+
+@threading_helper.requires_working_threading()
+class TestCBisect(unittest.TestCase, TestBase):
+    mod = c_bisect
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/Lib/test/test_free_threading/test_context.py b/Lib/test/test_free_threading/test_context.py
@@ -0,0 +1,57 @@
+import contextvars
+import unittest
+from threading import Event, Thread
+
+from test.support import threading_helper
+
+
+@threading_helper.requires_working_threading()
+class TestContext(unittest.TestCase):
+    def test_racing_read_write(self):
+        # gh-154535: reading a Context object from one thread while another
+        # thread sets variables in it used to crash.  The readers looked at
+        # Context.ctx_vars without owning a reference to it, so the writer
+        # could deallocate the mapping while a reader was walking it.
+        ctx = contextvars.Context()
+        cvars = [contextvars.ContextVar(f"cvar{i}") for i in range(64)]
+        done = Event()
+        errors = []
+
+        def writer():
+            def body():
+                i = 0
+                while not done.is_set():
+                    cvars[i % len(cvars)].set(i)
+                    i += 1
+            try:
+                ctx.run(body)
+            except BaseException as e:
+                errors.append(e)
+
+        def reader():
+            try:
+                for _ in range(200):
+                    ctx.copy()
+                    len(ctx)
+                    list(ctx)
+                    list(ctx.items())
+                    list(ctx.keys())
+                    list(ctx.values())
+                    cvars[0] in ctx
+                    ctx.get(cvars[0])
+                    ctx == ctx
+            except BaseException as e:
+                errors.append(e)
+            finally:
+                done.set()
+
+        threads = [Thread(target=writer)]
+        threads += [Thread(target=reader) for _ in range(4)]
+        with threading_helper.start_threads(threads, done.set):
+            pass
+
+        self.assertEqual(errors, [], msg=f"unexpected errors: {errors}")
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/Lib/test/test_free_threading/test_enumerate.py b/Lib/test/test_free_threading/test_enumerate.py
@@ -0,0 +1,38 @@
+import unittest
+import sys
+from threading import Thread, Barrier
+
+from test.support import threading_helper
+
+threading_helper.requires_working_threading(module=True)
+
+class EnumerateThreading(unittest.TestCase):
+
+    @threading_helper.reap_threads
+    def test_threading(self):
+        number_of_threads = 10
+        number_of_iterations = 8
+        n = 100
+        start = sys.maxsize - 40
+        barrier = Barrier(number_of_threads)
+        def work(enum):
+            barrier.wait()
+            while True:
+                try:
+                    _ = next(enum)
+                except StopIteration:
+                    break
+
+        for it in range(number_of_iterations):
+            enum = enumerate(tuple(range(start, start + n)))
+            worker_threads = []
+            for ii in range(number_of_threads):
+                worker_threads.append(
+                    Thread(target=work, args=[enum]))
+            with threading_helper.start_threads(worker_threads):
+                pass
+
+            barrier.reset()
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/Lib/test/test_free_threading/test_exceptions.py b/Lib/test/test_free_threading/test_exceptions.py
@@ -0,0 +1,16 @@
+import unittest
+import copy
+
+from test.support import threading_helper
+
+threading_helper.requires_working_threading(module=True)
+class ExceptionTests(unittest.TestCase):
+    def test_setstate_data_race(self):
+        E = Exception()
+
+        def func():
+            for i in range(100):
+                setattr(E, 'x', i)
+                copy.copy(E)
+
+        threading_helper.run_concurrently(func, nthreads=4)
diff --git a/Lib/test/test_free_threading/test_func_annotations.py b/Lib/test/test_free_threading/test_func_annotations.py
@@ -0,0 +1,67 @@
+import concurrent.futures
+import unittest
+import inspect
+from threading import Barrier
+from unittest import TestCase
+
+from test.support import threading_helper, Py_GIL_DISABLED
+
+threading_helper.requires_working_threading(module=True)
+
+
+def get_func_annotation(f, b):
+    b.wait()
+    return inspect.get_annotations(f)
+
+
+def get_func_annotation_dunder(f, b):
+    b.wait()
+    return f.__annotations__
+
+
+def set_func_annotation(f, b):
+    b.wait()
+    f.__annotations__ = {'x': int, 'y': int, 'return': int}
+    return f.__annotations__
+
+
+@unittest.skipUnless(Py_GIL_DISABLED, "Enable only in FT build")
+class TestFTFuncAnnotations(TestCase):
+    NUM_THREADS = 4
+
+    def test_concurrent_read(self):
+        def f(x: int) -> int:
+            return x + 1
+
+        for _ in range(10):
+            with concurrent.futures.ThreadPoolExecutor(max_workers=self.NUM_THREADS) as executor:
+                b = Barrier(self.NUM_THREADS)
+                futures = {executor.submit(get_func_annotation, f, b): i for i in range(self.NUM_THREADS)}
+                for fut in concurrent.futures.as_completed(futures):
+                    annotate = fut.result()
+                    self.assertIsNotNone(annotate)
+                    self.assertEqual(annotate, {'x': int, 'return': int})
+
+            with concurrent.futures.ThreadPoolExecutor(max_workers=self.NUM_THREADS) as executor:
+                b = Barrier(self.NUM_THREADS)
+                futures = {executor.submit(get_func_annotation_dunder, f, b): i for i in range(self.NUM_THREADS)}
+                for fut in concurrent.futures.as_completed(futures):
+                    annotate = fut.result()
+                    self.assertIsNotNone(annotate)
+                    self.assertEqual(annotate, {'x': int, 'return': int})
+
+    def test_concurrent_write(self):
+        def bar(x: int, y: float) -> float:
+            return y ** x
+
+        for _ in range(10):
+            with concurrent.futures.ThreadPoolExecutor(max_workers=self.NUM_THREADS) as executor:
+                b = Barrier(self.NUM_THREADS)
+                futures = {executor.submit(set_func_annotation, bar, b): i for i in range(self.NUM_THREADS)}
+                for fut in concurrent.futures.as_completed(futures):
+                    annotate = fut.result()
+                    self.assertIsNotNone(annotate)
+                    self.assertEqual(annotate, {'x': int, 'y': int, 'return': int})
+
+            # func_get_annotations returns in-place dict, so bar.__annotations__ should be modified as well
+            self.assertEqual(bar.__annotations__, {'x': int, 'y': int, 'return': int})
diff --git a/Lib/test/test_free_threading/test_functions.py b/Lib/test/test_free_threading/test_functions.py
@@ -0,0 +1,104 @@
+import random
+import unittest
+from unittest import TestCase
+
+from test.support import threading_helper
+
+threading_helper.requires_working_threading(module=True)
+
+NUM_THREADS = 8
+ITERS = 200
+
+
+def random_string():
+    return ''.join(random.choice('0123456789ABCDEF') for _ in range(10))
+
+
+def template_a(): pass
+def template_b(): pass
+
+
+class TestFTFunctionAttributes(TestCase):
+
+    def stress_attribute(self, attr, make_value):
+        def target(x=1):
+            return x
+
+        def writer():
+            for _ in range(ITERS):
+                setattr(target, attr, make_value())
+                getattr(target, attr)
+
+        threading_helper.run_concurrently(writer, NUM_THREADS)
+
+    def test_name(self):
+        self.stress_attribute("__name__", random_string)
+
+    def test_qualname(self):
+        self.stress_attribute("__qualname__", random_string)
+
+    def test_code(self):
+        codes = (template_a.__code__, template_b.__code__)
+        self.stress_attribute("__code__", lambda: random.choice(codes))
+
+    def test_defaults(self):
+        self.stress_attribute("__defaults__", lambda: (random_string(),))
+
+    def test_kwdefaults(self):
+        self.stress_attribute("__kwdefaults__", lambda: {"x": random_string()})
+
+    def test_annotations(self):
+        self.stress_attribute("__annotations__",
+                              lambda: {"x": random_string()})
+
+    def test_annotate(self):
+        self.stress_attribute("__annotate__",
+                              lambda: (lambda format: {"x": str}))
+
+    def test_type_params(self):
+        self.stress_attribute("__type_params__", lambda: (random_string(),))
+
+    def test_doc(self):
+        self.stress_attribute("__doc__", random_string)
+
+    def test_module(self):
+        self.stress_attribute("__module__", random_string)
+
+    def test_annotations_and_annotate(self):
+        # The __annotations__ and __annotate__ setters clear each other.
+        def target(): pass
+
+        def set_annotations():
+            for _ in range(ITERS):
+                target.__annotations__ = {"x": random_string()}
+                target.__annotations__
+
+        def set_annotate():
+            for _ in range(ITERS):
+                target.__annotate__ = lambda format: {"x": str}
+                target.__annotate__
+
+        threading_helper.run_concurrently(
+            [set_annotations, set_annotate] * (NUM_THREADS // 2))
+
+    def test_call_while_replacing_defaults(self):
+        # The eval loop reads __defaults__ and __kwdefaults__ without holding
+        # a lock while pushing a frame.
+        def target(x="init", *, y="init"):
+            return x, y
+
+        def writer():
+            for _ in range(ITERS):
+                target.__defaults__ = (random_string(),)
+                target.__kwdefaults__ = {"y": random_string()}
+
+        def caller():
+            for _ in range(ITERS):
+                target()
+
+        threading_helper.run_concurrently(
+            [writer, caller] * (NUM_THREADS // 2))
+
+
+if __name__ == "__main__":
+    unittest.main()