youknowone · Oct 29, 2026 · Oct 29, 2026
diff --git a/Lib/test/test_free_threading/test_cprofile.py b/Lib/test/test_free_threading/test_cprofile.py
@@ -0,0 +1,43 @@
+import unittest
+
+from test.support import threading_helper
+
+import cProfile
+import pstats
+
+
+NTHREADS = 10
+INSERT_PER_THREAD = 1000
+
+
+@threading_helper.requires_working_threading()
+class TestCProfile(unittest.TestCase):
+    def test_cprofile_racing_list_insert(self):
+        def list_insert(lst):
+            for i in range(INSERT_PER_THREAD):
+                lst.insert(0, i)
+
+        lst = []
+
+        with cProfile.Profile() as pr:
+            threading_helper.run_concurrently(
+                worker_func=list_insert, nthreads=NTHREADS, args=(lst,)
+            )
+            pr.create_stats()
+            ps = pstats.Stats(pr)
+            stats_profile = ps.get_stats_profile()
+            list_insert_profile = stats_profile.func_profiles[
+                "<method 'insert' of 'list' objects>"
+            ]
+            # Even though there is no explicit recursive call to insert,
+            # cProfile may record some calls as recursive due to limitations
+            # in its handling of multithreaded programs. This issue is not
+            # directly related to FT Python itself; however, it tends to be
+            # more noticeable when using FT Python. Therefore, consider only
+            # the calls section and disregard the recursive part.
+            list_insert_ncalls = list_insert_profile.ncalls.split("/")[0]
+            self.assertEqual(
+                int(list_insert_ncalls), NTHREADS * INSERT_PER_THREAD
+            )
+
+        self.assertEqual(len(lst), NTHREADS * INSERT_PER_THREAD)
diff --git a/Lib/test/test_free_threading/test_io.py b/Lib/test/test_free_threading/test_io.py
@@ -0,0 +1,200 @@
+import codecs
+import io
+import _pyio as pyio
+import threading
+from unittest import TestCase
+from test.support import threading_helper
+from test.support.threading_helper import run_concurrently
+from random import randint
+from sys import getsizeof
+
+threading_helper.requires_working_threading(module=True)
+
+
+class ThreadSafetyMixin:
+    # Test pretty much everything that can break under free-threading.
+    # Non-deterministic, but at least one of these things will fail if
+    # BytesIO object is not free-thread safe.
+
+    def check(self, funcs, *args):
+        barrier = threading.Barrier(len(funcs))
+        threads = []
+
+        for func in funcs:
+            thread = threading.Thread(target=func, args=(barrier, *args))
+
+            threads.append(thread)
+
+        with threading_helper.start_threads(threads):
+            pass
+
+    @threading_helper.requires_working_threading()
+    @threading_helper.reap_threads
+    def test_free_threading(self):
+        """Test for segfaults and aborts."""
+
+        def write(barrier, b, *ignore):
+            barrier.wait()
+            try: b.write(b'0' * randint(100, 1000))
+            except ValueError: pass  # ignore write fail to closed file
+
+        def writelines(barrier, b, *ignore):
+            barrier.wait()
+            b.write(b'0\n' * randint(100, 1000))
+
+        def truncate(barrier, b, *ignore):
+            barrier.wait()
+            try: b.truncate(0)
+            except BufferError: pass  # ignore exported buffer
+
+        def read(barrier, b, *ignore):
+            barrier.wait()
+            b.read()
+
+        def read1(barrier, b, *ignore):
+            barrier.wait()
+            b.read1()
+
+        def readline(barrier, b, *ignore):
+            barrier.wait()
+            b.readline()
+
+        def readlines(barrier, b, *ignore):
+            barrier.wait()
+            b.readlines()
+
+        def readinto(barrier, b, into, *ignore):
+            barrier.wait()
+            b.readinto(into)
+
+        def close(barrier, b, *ignore):
+            barrier.wait()
+            b.close()
+
+        def getvalue(barrier, b, *ignore):
+            barrier.wait()
+            b.getvalue()
+
+        def getbuffer(barrier, b, *ignore):
+            barrier.wait()
+            b.getbuffer()
+
+        def iter(barrier, b, *ignore):
+            barrier.wait()
+            list(b)
+
+        def getstate(barrier, b, *ignore):
+            barrier.wait()
+            b.__getstate__()
+
+        def setstate(barrier, b, st, *ignore):
+            barrier.wait()
+            b.__setstate__(st)
+
+        def sizeof(barrier, b, *ignore):
+            barrier.wait()
+            getsizeof(b)
+
+        self.check([write] * 10, self.ioclass())
+        self.check([writelines] * 10, self.ioclass())
+        self.check([write] * 10 + [truncate] * 10, self.ioclass())
+        self.check([truncate] + [read] * 10, self.ioclass(b'0\n'*204800))
+        self.check([truncate] + [read1] * 10, self.ioclass(b'0\n'*204800))
+        self.check([truncate] + [readline] * 10, self.ioclass(b'0\n'*20480))
+        self.check([truncate] + [readlines] * 10, self.ioclass(b'0\n'*20480))
+        self.check([truncate] + [readinto] * 10, self.ioclass(b'0\n'*204800), bytearray(b'0\n'*204800))
+        self.check([close] + [write] * 10, self.ioclass())
+        self.check([truncate] + [getvalue] * 10, self.ioclass(b'0\n'*204800))
+        self.check([truncate] + [getbuffer] * 10, self.ioclass(b'0\n'*204800))
+        self.check([truncate] + [iter] * 10, self.ioclass(b'0\n'*20480))
+        self.check([truncate] + [getstate] * 10, self.ioclass(b'0\n'*204800))
+        state = self.ioclass(b'123').__getstate__()
+        self.check([truncate] + [setstate] * 10, self.ioclass(b'0\n'*204800), state)
+        self.check([truncate] + [sizeof] * 10, self.ioclass(b'0\n'*204800))
+
+        # no tests for seek or tell because they don't break anything
+
+class CBytesIOTest(ThreadSafetyMixin, TestCase):
+    ioclass = io.BytesIO
+
+    @threading_helper.requires_working_threading()
+    @threading_helper.reap_threads
+    def test_concurrent_whole_buffer_read_and_resize(self):
+        shared = self.ioclass(b"x" * 64)
+        writers = 2
+        readers = 8
+        loops = 2000
+        barrier = threading.Barrier(writers + readers)
+
+        def writer():
+            barrier.wait()
+            for i in range(loops):
+                shared.seek(0)
+                shared.write(b"a" * (64 + (i & 63)))
+
+        def reader():
+            barrier.wait()
+            for _ in range(loops):
+                shared.seek(0)
+                shared.read()
+                shared.seek(0)
+                shared.getvalue()
+
+        threads = [threading.Thread(target=writer) for _ in range(writers)]
+        threads += [threading.Thread(target=reader) for _ in range(readers)]
+        with threading_helper.start_threads(threads):
+            pass
+
+class PyBytesIOTest(ThreadSafetyMixin, TestCase):
+     ioclass = pyio.BytesIO
+
+
+class IncrementalNewlineDecoderTest(TestCase):
+    def make_decoder(self):
+        utf8_decoder = codecs.getincrementaldecoder('utf-8')()
+        return io.IncrementalNewlineDecoder(utf8_decoder, translate=True)
+
+    def test_concurrent_reset(self):
+        decoder = self.make_decoder()
+
+        def worker():
+            for _ in range(100):
+                decoder.reset()
+
+        run_concurrently(worker_func=worker, nthreads=2)
+
+    def test_concurrent_decode(self):
+        decoder = self.make_decoder()
+
+        def worker():
+            for _ in range(100):
+                decoder.decode(b"line\r\n", final=False)
+
+        run_concurrently(worker_func=worker, nthreads=2)
+
+    def test_concurrent_getstate_setstate(self):
+        decoder = self.make_decoder()
+        state = decoder.getstate()
+
+        def getstate_worker():
+            for _ in range(100):
+                decoder.getstate()
+
+        def setstate_worker():
+            for _ in range(100):
+                decoder.setstate(state)
+
+        run_concurrently([getstate_worker] * 2 + [setstate_worker] * 2)
+
+    def test_concurrent_decode_and_reset(self):
+        decoder = self.make_decoder()
+
+        def decode_worker():
+            for _ in range(100):
+                decoder.decode(b"line\r\n", final=False)
+
+        def reset_worker():
+            for _ in range(100):
+                decoder.reset()
+
+        run_concurrently([decode_worker] * 2 + [reset_worker] * 2)
diff --git a/Lib/test/test_free_threading/test_iteration.py b/Lib/test/test_free_threading/test_iteration.py
@@ -0,0 +1,127 @@
+import threading
+import unittest
+from test import support
+
+# The race conditions these tests were written for only happen every now and
+# then, even with the current numbers. To find rare race conditions, bumping
+# these up will help, but it makes the test runtime highly variable under
+# free-threading. Overhead is much higher under ThreadSanitizer, but it's
+# also much better at detecting certain races, so we don't need as many
+# items/threads.
+if support.check_sanitizer(thread=True):
+    NUMITEMS = 1000
+    NUMTHREADS = 2
+else:
+    NUMITEMS = 5000
+    NUMTHREADS = 5
+NUMMUTATORS = 2
+
+class ContendedTupleIterationTest(unittest.TestCase):
+    def make_testdata(self, n):
+        return tuple(range(n))
+
+    def assert_iterator_results(self, results, expected):
+        # Most iterators are not atomic (yet?) so they can skip or duplicate
+        # items, but they should not invent new items (like the range
+        # iterator currently does).
+        extra_items = set(results) - set(expected)
+        self.assertEqual(set(), extra_items)
+
+    def run_threads(self, func, *args, numthreads=NUMTHREADS):
+        threads = []
+        for _ in range(numthreads):
+            t = threading.Thread(target=func, args=args)
+            t.start()
+            threads.append(t)
+        return threads
+
+    def test_iteration(self):
+        """Test iteration over a shared container"""
+        seq = self.make_testdata(NUMITEMS)
+        results = []
+        start = threading.Barrier(NUMTHREADS)
+        def worker():
+            idx = 0
+            start.wait()
+            for item in seq:
+                idx += 1
+            results.append(idx)
+        threads = self.run_threads(worker)
+        for t in threads:
+            t.join()
+        # Each thread has its own iterator, so results should be entirely predictable.
+        self.assertEqual(results, [NUMITEMS] * NUMTHREADS)
+
+    def test_shared_iterator(self):
+        """Test iteration over a shared iterator"""
+        seq = self.make_testdata(NUMITEMS)
+        it = iter(seq)
+        results = []
+        start = threading.Barrier(NUMTHREADS)
+        def worker():
+            items = []
+            start.wait()
+            # We want a tight loop, so put items in the shared list at the end.
+            for item in it:
+                items.append(item)
+            results.extend(items)
+        threads = self.run_threads(worker)
+        for t in threads:
+            t.join()
+        self.assert_iterator_results(results, seq)
+
+class ContendedListIterationTest(ContendedTupleIterationTest):
+    def make_testdata(self, n):
+        return list(range(n))
+
+    def test_iteration_while_mutating(self):
+        """Test iteration over a shared mutating container."""
+        seq = self.make_testdata(NUMITEMS)
+        results = []
+        start = threading.Barrier(NUMTHREADS + NUMMUTATORS)
+        endmutate = threading.Event()
+        def mutator():
+            orig = seq[:]
+            # Make changes big enough to cause resizing of the list, with
+            # items shifted around for good measure.
+            replacement = (orig * 3)[NUMITEMS//2:]
+            start.wait()
+            while not endmutate.is_set():
+                seq.extend(replacement)
+                seq[:0] = orig
+                seq.__imul__(2)
+                seq.extend(seq)
+                seq[:] = orig
+        def worker():
+            items = []
+            start.wait()
+            # We want a tight loop, so put items in the shared list at the end.
+            for item in seq:
+                items.append(item)
+            results.extend(items)
+        mutators = ()
+        try:
+            threads = self.run_threads(worker)
+            mutators = self.run_threads(mutator, numthreads=NUMMUTATORS)
+            for t in threads:
+                t.join()
+        finally:
+            endmutate.set()
+            for m in mutators:
+                m.join()
+        self.assert_iterator_results(results, list(seq))
+
+
+class ContendedRangeIterationTest(ContendedTupleIterationTest):
+    def make_testdata(self, n):
+        return range(n)
+
+    def assert_iterator_results(self, results, expected):
+        # Range iterators that are shared between threads will (right now)
+        # sometimes produce items after the end of the range, sometimes
+        # _far_ after the end of the range. That should be fixed, but for
+        # now, let's just check they're integers that could have resulted
+        # from stepping beyond the range bounds.
+        extra_items = set(results) - set(expected)
+        for item in extra_items:
+            self.assertEqual((item - expected.start) % expected.step, 0)