From e889fa166540f39676370985447bf0b761447c0c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?tonghuaroot=20=28=E7=AB=A5=E8=AF=9D=29?= Date: Fri, 10 Jul 2026 10:38:58 +0800 Subject: [PATCH] gh-153062: Fix a crash iterating itertools.tee on the free-threaded build (gh-153063) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit itertools.tee branches share a linked list of teedataobject cells. On the free-threaded build, iterating one branch from multiple threads, or iterating sibling branches concurrently, raced on the shared cells and each branch's position, corrupting refcounts and crashing. Lock each teedataobject while reading, extending, or clearing it, and snapshot each branch's position under the tee object's own lock, revalidating before advancing, so the two locks are never nested. Concurrent iteration of one tee is undefined and may raise RuntimeError as documented, but no longer crashes. The free-threading snapshot and revalidation add per-element overhead on the default build, where the GIL already serializes access. Guard that path under Py_GIL_DISABLED so the default build keeps the original iteration and the free-threaded path is unchanged. (cherry picked from commit 6c389c446e487bcb9d09e85de30aa42052990f1c) Co-authored-by: tonghuaroot (童话) Co-authored-by: Neil Schemenauer --- .../test_free_threading/test_itertools.py | 88 +++++++++++++++++-- ...-07-05-14-30-00.gh-issue-153062.teeFT1.rst | 2 + Modules/itertoolsmodule.c | 82 +++++++++++++++-- 3 files changed, 157 insertions(+), 15 deletions(-) create mode 100644 Misc/NEWS.d/next/Library/2026-07-05-14-30-00.gh-issue-153062.teeFT1.rst diff --git a/Lib/test/test_free_threading/test_itertools.py b/Lib/test/test_free_threading/test_itertools.py index 670d4ca8835e0d2..7392bd739acd709 100644 --- a/Lib/test/test_free_threading/test_itertools.py +++ b/Lib/test/test_free_threading/test_itertools.py @@ -1,5 +1,14 @@ import unittest -from itertools import accumulate, batched, chain, combinations_with_replacement, cycle, permutations, zip_longest +from itertools import ( + accumulate, + batched, + chain, + combinations_with_replacement, + cycle, + permutations, + tee, + zip_longest, +) from test.support import threading_helper @@ -15,20 +24,23 @@ def work_iterator(it): class ItertoolsThreading(unittest.TestCase): - @threading_helper.reap_threads def test_accumulate(self): number_of_iterations = 10 for _ in range(number_of_iterations): it = accumulate(tuple(range(40))) - threading_helper.run_concurrently(work_iterator, nthreads=10, args=[it]) + threading_helper.run_concurrently( + work_iterator, nthreads=10, args=[it] + ) @threading_helper.reap_threads def test_batched(self): number_of_iterations = 10 for _ in range(number_of_iterations): it = batched(tuple(range(1000)), 2) - threading_helper.run_concurrently(work_iterator, nthreads=10, args=[it]) + threading_helper.run_concurrently( + work_iterator, nthreads=10, args=[it] + ) @threading_helper.reap_threads def test_cycle(self): @@ -46,28 +58,88 @@ def test_chain(self): number_of_iterations = 10 for _ in range(number_of_iterations): it = chain(*[(1,)] * 200) - threading_helper.run_concurrently(work_iterator, nthreads=6, args=[it]) + threading_helper.run_concurrently( + work_iterator, nthreads=6, args=[it] + ) @threading_helper.reap_threads def test_combinations_with_replacement(self): number_of_iterations = 6 for _ in range(number_of_iterations): it = combinations_with_replacement(tuple(range(2)), 2) - threading_helper.run_concurrently(work_iterator, nthreads=6, args=[it]) + threading_helper.run_concurrently( + work_iterator, nthreads=6, args=[it] + ) @threading_helper.reap_threads def test_permutations(self): number_of_iterations = 6 for _ in range(number_of_iterations): it = permutations(tuple(range(4)), 2) - threading_helper.run_concurrently(work_iterator, nthreads=6, args=[it]) + threading_helper.run_concurrently( + work_iterator, nthreads=6, args=[it] + ) @threading_helper.reap_threads def test_zip_longest(self): number_of_iterations = 10 for _ in range(number_of_iterations): it = zip_longest(list(range(4)), list(range(8)), fillvalue=0) - threading_helper.run_concurrently(work_iterator, nthreads=10, args=[it]) + threading_helper.run_concurrently( + work_iterator, nthreads=10, args=[it] + ) + + +class TestTeeConcurrent(unittest.TestCase): + # itertools.tee branches share a linked list of internal data cells. + # Concurrent iteration must not corrupt that shared state or crash the + # free-threaded build. A crash shows up as the interpreter dying (not as a + # caught exception); tee is documented as not thread-safe, so a + # ``RuntimeError`` from the re-entrancy guard is an allowed outcome and is + # tolerated here. + + def test_same_branch(self): + # Many threads consume the same tee branch. + errors = [] + + def consume(it): + try: + for _ in it: + pass + except RuntimeError: + pass + except Exception as e: + errors.append(e) + + for _ in range(100): + a, _ = tee(iter(range(2000)), 2) + threading_helper.run_concurrently(consume, nthreads=8, args=(a,)) + + self.assertEqual(errors, [], msg=f"unexpected errors: {errors}") + + def test_sibling_branches(self): + # Each thread consumes a different sibling branch of the same tee. + errors = [] + + def make_worker(it): + def consume(): + try: + for _ in it: + pass + except RuntimeError: + pass + except Exception as e: + errors.append(e) + + return consume + + for _ in range(100): + branches = tee(iter(range(4000)), 8) + threading_helper.run_concurrently( + [make_worker(it) for it in branches] + ) + + self.assertEqual(errors, [], msg=f"unexpected errors: {errors}") if __name__ == "__main__": diff --git a/Misc/NEWS.d/next/Library/2026-07-05-14-30-00.gh-issue-153062.teeFT1.rst b/Misc/NEWS.d/next/Library/2026-07-05-14-30-00.gh-issue-153062.teeFT1.rst new file mode 100644 index 000000000000000..ed1aaa15d37bd9c --- /dev/null +++ b/Misc/NEWS.d/next/Library/2026-07-05-14-30-00.gh-issue-153062.teeFT1.rst @@ -0,0 +1,2 @@ +Fix a crash when concurrently iterating an :func:`itertools.tee` iterator on +the free-threaded build. diff --git a/Modules/itertoolsmodule.c b/Modules/itertoolsmodule.c index 0dd31dfbc5a3469..72bfab1abaf9cae 100644 --- a/Modules/itertoolsmodule.c +++ b/Modules/itertoolsmodule.c @@ -768,13 +768,17 @@ teedataobject_newinternal(itertools_state *state, PyObject *it) static PyObject * teedataobject_jumplink(itertools_state *state, teedataobject *tdo) { + PyObject *link; + Py_BEGIN_CRITICAL_SECTION(tdo); if (tdo->nextlink == NULL) tdo->nextlink = teedataobject_newinternal(state, tdo->it); - return Py_XNewRef(tdo->nextlink); + link = Py_XNewRef(tdo->nextlink); + Py_END_CRITICAL_SECTION(); + return link; } static PyObject * -teedataobject_getitem(teedataobject *tdo, int i) +teedataobject_getitem_lock_held(teedataobject *tdo, int i) { PyObject *value; @@ -800,6 +804,16 @@ teedataobject_getitem(teedataobject *tdo, int i) return Py_NewRef(value); } +static PyObject * +teedataobject_getitem(teedataobject *tdo, int i) +{ + PyObject *result; + Py_BEGIN_CRITICAL_SECTION(tdo); + result = teedataobject_getitem_lock_held(tdo, i); + Py_END_CRITICAL_SECTION(); + return result; +} + static int teedataobject_traverse(PyObject *op, visitproc visit, void * arg) { @@ -819,8 +833,11 @@ teedataobject_safe_decref(PyObject *obj) { while (obj && _PyObject_IsUniquelyReferenced(obj)) { teedataobject *tmp = teedataobject_CAST(obj); - PyObject *nextlink = tmp->nextlink; + PyObject *nextlink; + Py_BEGIN_CRITICAL_SECTION(obj); + nextlink = tmp->nextlink; tmp->nextlink = NULL; + Py_END_CRITICAL_SECTION(); Py_SETREF(obj, nextlink); } Py_XDECREF(obj); @@ -833,11 +850,13 @@ teedataobject_clear(PyObject *op) PyObject *tmp; teedataobject *tdo = teedataobject_CAST(op); + Py_BEGIN_CRITICAL_SECTION(op); Py_CLEAR(tdo->it); for (i=0 ; inumread ; i++) Py_CLEAR(tdo->values[i]); tmp = tdo->nextlink; tdo->nextlink = NULL; + Py_END_CRITICAL_SECTION(); teedataobject_safe_decref(tmp); return 0; } @@ -930,20 +949,67 @@ static PyObject * tee_next(PyObject *op) { teeobject *to = teeobject_CAST(op); - PyObject *value, *link; + PyObject *value; +#ifndef Py_GIL_DISABLED + /* The GIL already serializes access, so keep the simple path without the + snapshot and revalidation that the free-threaded build needs. */ if (to->index >= LINKCELLS) { - link = teedataobject_jumplink(to->state, to->dataobj); - if (link == NULL) + PyObject *link = teedataobject_jumplink(to->state, to->dataobj); + if (link == NULL) { return NULL; + } Py_SETREF(to->dataobj, (teedataobject *)link); to->index = 0; } value = teedataobject_getitem(to->dataobj, to->index); - if (value == NULL) + if (value == NULL) { return NULL; + } to->index++; return value; +#else + for (;;) { + teedataobject *dataobj; + int index; + + /* Snapshot the branch position (strong ref to the shared data object) + under the tee lock; the data object is locked separately, not nested, + then the advance is revalidated. */ + Py_BEGIN_CRITICAL_SECTION(op); + dataobj = (teedataobject *)Py_NewRef((PyObject *)to->dataobj); + index = to->index; + Py_END_CRITICAL_SECTION(); + + if (index < LINKCELLS) { + value = teedataobject_getitem(dataobj, index); + if (value != NULL) { + Py_BEGIN_CRITICAL_SECTION(op); + if (to->dataobj == dataobj && to->index == index) { + to->index = index + 1; + } + Py_END_CRITICAL_SECTION(); + } + Py_DECREF(dataobj); + return value; + } + + PyObject *link = teedataobject_jumplink(to->state, dataobj); + if (link == NULL) { + Py_DECREF(dataobj); + return NULL; + } + Py_BEGIN_CRITICAL_SECTION(op); + if (to->dataobj == dataobj) { + Py_SETREF(to->dataobj, (teedataobject *)link); + to->index = 0; + link = NULL; + } + Py_END_CRITICAL_SECTION(); + Py_XDECREF(link); + Py_DECREF(dataobj); + } +#endif } static int @@ -962,8 +1028,10 @@ tee_copy_impl(teeobject *to) if (newto == NULL) { return NULL; } + Py_BEGIN_CRITICAL_SECTION(to); newto->dataobj = (teedataobject *)Py_NewRef(to->dataobj); newto->index = to->index; + Py_END_CRITICAL_SECTION(); newto->weakreflist = NULL; newto->state = to->state; PyObject_GC_Track(newto);