diff --git a/Lib/test/test_free_threading/test_itertools.py b/Lib/test/test_free_threading/test_itertools.py index 2d2d4ff1ca1f6f1..0703c7e5e2916e8 100644 --- a/Lib/test/test_free_threading/test_itertools.py +++ b/Lib/test/test_free_threading/test_itertools.py @@ -1,6 +1,13 @@ import unittest +from threading import Thread, Barrier from itertools import ( + accumulate, + chain, + combinations_with_replacement, + cycle, + permutations, tee, + zip_longest, ) from test.support import threading_helper @@ -8,6 +15,98 @@ threading_helper.requires_working_threading(module=True) +def work_iterator(it): + while True: + try: + next(it) + except StopIteration: + break + + +class ItertoolsThreading(unittest.TestCase): + + @threading_helper.reap_threads + def test_cycle(self): + number_of_threads = 6 + number_of_iterations = 10 + number_of_cycles = 400 + + barrier = Barrier(number_of_threads) + def work(it): + barrier.wait() + for _ in range(number_of_cycles): + _ = next(it) + + data = (1, 2, 3, 4) + for it in range(number_of_iterations): + cycle_iterator = cycle(data) + worker_threads = [] + for ii in range(number_of_threads): + worker_threads.append( + Thread(target=work, args=[cycle_iterator])) + + with threading_helper.start_threads(worker_threads): + pass + + barrier.reset() + + @threading_helper.reap_threads + def test_chain(self): + number_of_threads = 6 + number_of_iterations = 20 + + barrier = Barrier(number_of_threads) + def work(it): + barrier.wait() + while True: + try: + next(it) + except StopIteration: + break + + data = [(1, )] * 200 + for it in range(number_of_iterations): + chain_iterator = chain(*data) + worker_threads = [] + for ii in range(number_of_threads): + worker_threads.append( + Thread(target=work, args=[chain_iterator])) + + with threading_helper.start_threads(worker_threads): + pass + + barrier.reset() + + @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.reap_threads + def test_permutations(self): + number_of_iterations = 6 + for _ in range(number_of_iterations): + it = permutations(tuple(range(2)), 2) + threading_helper.run_concurrently(work_iterator, nthreads=6, args=[it]) + + @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.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]) + + + 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 diff --git a/Lib/test/test_free_threading/test_itertools_combinatoric.py b/Lib/test/test_free_threading/test_itertools_combinatoric.py new file mode 100644 index 000000000000000..5b3b88deedd1210 --- /dev/null +++ b/Lib/test/test_free_threading/test_itertools_combinatoric.py @@ -0,0 +1,51 @@ +import unittest +from threading import Thread, Barrier +from itertools import combinations, product +from test.support import threading_helper + + +threading_helper.requires_working_threading(module=True) + +def test_concurrent_iteration(iterator, number_of_threads): + barrier = Barrier(number_of_threads) + def iterator_worker(it): + barrier.wait() + while True: + try: + _ = next(it) + except StopIteration: + return + + worker_threads = [] + for ii in range(number_of_threads): + worker_threads.append( + Thread(target=iterator_worker, args=[iterator])) + + with threading_helper.start_threads(worker_threads): + pass + + barrier.reset() + +class ItertoolsThreading(unittest.TestCase): + + @threading_helper.reap_threads + def test_combinations(self): + number_of_threads = 10 + number_of_iterations = 24 + + for it in range(number_of_iterations): + iterator = combinations((1, 2, 3, 4, 5), 2) + test_concurrent_iteration(iterator, number_of_threads) + + @threading_helper.reap_threads + def test_product(self): + number_of_threads = 10 + number_of_iterations = 24 + + for it in range(number_of_iterations): + iterator = product((1, 2, 3, 4, 5), (10, 20, 30)) + test_concurrent_iteration(iterator, number_of_threads) + + +if __name__ == "__main__": + unittest.main() diff --git a/Misc/NEWS.d/next/Library/2025-03-13-20-48-58.gh-issue-123471.cM4w4f.rst b/Misc/NEWS.d/next/Library/2025-03-13-20-48-58.gh-issue-123471.cM4w4f.rst new file mode 100644 index 000000000000000..cfc783900de70fe --- /dev/null +++ b/Misc/NEWS.d/next/Library/2025-03-13-20-48-58.gh-issue-123471.cM4w4f.rst @@ -0,0 +1 @@ +Make concurrent iterations over :class:`itertools.cycle` safe under free-threading. diff --git a/Misc/NEWS.d/next/Library/2025-04-22-21-00-23.gh-issue-123471.asOLA2.rst b/Misc/NEWS.d/next/Library/2025-04-22-21-00-23.gh-issue-123471.asOLA2.rst new file mode 100644 index 000000000000000..a4b4b6d2c23d495 --- /dev/null +++ b/Misc/NEWS.d/next/Library/2025-04-22-21-00-23.gh-issue-123471.asOLA2.rst @@ -0,0 +1 @@ +Make concurrent iterations over :class:`itertools.combinations` and :class:`itertools.product` safe under free-threading. diff --git a/Misc/NEWS.d/next/Library/2025-06-18-19-25-32.gh-issue-123471.lx1Xbt.rst b/Misc/NEWS.d/next/Library/2025-06-18-19-25-32.gh-issue-123471.lx1Xbt.rst new file mode 100644 index 000000000000000..6f395024a9e1790 --- /dev/null +++ b/Misc/NEWS.d/next/Library/2025-06-18-19-25-32.gh-issue-123471.lx1Xbt.rst @@ -0,0 +1 @@ +Make concurrent iterations over :class:`itertools.chain` safe under :term:`free threading`. diff --git a/Misc/NEWS.d/next/Library/2026-02-03-08-50-58.gh-issue-123471.yF1Gym.rst b/Misc/NEWS.d/next/Library/2026-02-03-08-50-58.gh-issue-123471.yF1Gym.rst new file mode 100644 index 000000000000000..85e9a03426e1fca --- /dev/null +++ b/Misc/NEWS.d/next/Library/2026-02-03-08-50-58.gh-issue-123471.yF1Gym.rst @@ -0,0 +1 @@ +Make concurrent iteration over :class:`itertools.combinations_with_replacement` and :class:`itertools.permutations` safe under free-threading. diff --git a/Misc/NEWS.d/next/Library/2026-02-04-20-30-59.gh-issue-123471.1dnPvs.rst b/Misc/NEWS.d/next/Library/2026-02-04-20-30-59.gh-issue-123471.1dnPvs.rst new file mode 100644 index 000000000000000..d650103e28ee686 --- /dev/null +++ b/Misc/NEWS.d/next/Library/2026-02-04-20-30-59.gh-issue-123471.1dnPvs.rst @@ -0,0 +1 @@ +Make concurrent iteration over :class:`itertools.accumulate` safe under free-threading. diff --git a/Misc/NEWS.d/next/Library/2026-03-17-19-51-05.gh-issue-123471.oY4UR5.rst b/Misc/NEWS.d/next/Library/2026-03-17-19-51-05.gh-issue-123471.oY4UR5.rst new file mode 100644 index 000000000000000..8d2e1b970e81711 --- /dev/null +++ b/Misc/NEWS.d/next/Library/2026-03-17-19-51-05.gh-issue-123471.oY4UR5.rst @@ -0,0 +1 @@ +Make concurrent iteration over :class:`itertools.zip_longest` safe under free-threading. diff --git a/Modules/itertoolsmodule.c b/Modules/itertoolsmodule.c index ffe44595ac59e77..c307484d13377d4 100644 --- a/Modules/itertoolsmodule.c +++ b/Modules/itertoolsmodule.c @@ -1211,7 +1211,6 @@ typedef struct { PyObject *it; PyObject *saved; Py_ssize_t index; - int firstpass; } cycleobject; #define cycleobject_CAST(op) ((cycleobject *)(op)) @@ -1252,8 +1251,7 @@ itertools_cycle_impl(PyTypeObject *type, PyObject *iterable) } lz->it = it; lz->saved = saved; - lz->index = 0; - lz->firstpass = 0; + lz->index = -1; return (PyObject *)lz; } @@ -1286,11 +1284,11 @@ cycle_next(PyObject *op) cycleobject *lz = cycleobject_CAST(op); PyObject *item; - if (lz->it != NULL) { + Py_ssize_t index = FT_ATOMIC_LOAD_SSIZE_RELAXED(lz->index); + + if (index < 0) { item = PyIter_Next(lz->it); if (item != NULL) { - if (lz->firstpass) - return item; if (PyList_Append(lz->saved, item)) { Py_DECREF(item); return NULL; @@ -1300,15 +1298,22 @@ cycle_next(PyObject *op) /* Note: StopIteration is already cleared by PyIter_Next() */ if (PyErr_Occurred()) return NULL; + index = 0; + FT_ATOMIC_STORE_SSIZE_RELAXED(lz->index, 0); +#ifndef Py_GIL_DISABLED Py_CLEAR(lz->it); +#endif } if (PyList_GET_SIZE(lz->saved) == 0) return NULL; - item = PyList_GET_ITEM(lz->saved, lz->index); - lz->index++; - if (lz->index >= PyList_GET_SIZE(lz->saved)) - lz->index = 0; - return Py_NewRef(item); + item = PyList_GetItemRef(lz->saved, index); + assert(item); + index++; + if (index >= PyList_GET_SIZE(lz->saved)) { + index = 0; + } + FT_ATOMIC_STORE_SSIZE_RELAXED(lz->index, index); + return item; } static PyType_Slot cycle_slots[] = { @@ -1962,8 +1967,8 @@ chain_traverse(PyObject *op, visitproc visit, void *arg) return 0; } -static PyObject * -chain_next(PyObject *op) +static inline PyObject * +chain_next_lock_held(PyObject *op) { chainobject *lz = chainobject_CAST(op); PyObject *item; @@ -2001,6 +2006,16 @@ chain_next(PyObject *op) return NULL; } +static PyObject * +chain_next(PyObject *op) +{ + PyObject *result; + Py_BEGIN_CRITICAL_SECTION(op); + result = chain_next_lock_held(op); + Py_END_CRITICAL_SECTION() + return result; +} + PyDoc_STRVAR(chain_doc, "chain(*iterables)\n\ --\n\ @@ -2172,7 +2187,7 @@ product_traverse(PyObject *op, visitproc visit, void *arg) } static PyObject * -product_next(PyObject *op) +product_next_lock_held(PyObject *op) { productobject *lz = productobject_CAST(op); PyObject *pool; @@ -2258,6 +2273,16 @@ product_next(PyObject *op) return NULL; } +static PyObject * +product_next(PyObject *op) +{ + PyObject *result; + Py_BEGIN_CRITICAL_SECTION(op); + result = product_next_lock_held(op); + Py_END_CRITICAL_SECTION() + return result; +} + static PyMethodDef product_methods[] = { {"__sizeof__", product_sizeof, METH_NOARGS, sizeof_doc}, {NULL, NULL} /* sentinel */ @@ -2405,7 +2430,7 @@ combinations_traverse(PyObject *op, visitproc visit, void *arg) } static PyObject * -combinations_next(PyObject *op) +combinations_next_lock_held(PyObject *op) { combinationsobject *co = combinationsobject_CAST(op); PyObject *elem; @@ -2490,6 +2515,16 @@ combinations_next(PyObject *op) return NULL; } +static PyObject * +combinations_next(PyObject *op) +{ + PyObject *result; + Py_BEGIN_CRITICAL_SECTION(op); + result = combinations_next_lock_held(op); + Py_END_CRITICAL_SECTION() + return result; +} + static PyMethodDef combinations_methods[] = { {"__sizeof__", combinations_sizeof, METH_NOARGS, sizeof_doc}, {NULL, NULL} /* sentinel */ @@ -2649,7 +2684,7 @@ cwr_traverse(PyObject *op, visitproc visit, void *arg) } static PyObject * -cwr_next(PyObject *op) +cwr_next_lock_held(PyObject *op) { cwrobject *co = cwrobject_CAST(op); PyObject *elem; @@ -2728,6 +2763,16 @@ cwr_next(PyObject *op) return NULL; } +static PyObject * +cwr_next(PyObject *op) +{ + PyObject *result; + Py_BEGIN_CRITICAL_SECTION(op); + result = cwr_next_lock_held(op); + Py_END_CRITICAL_SECTION() + return result; +} + static PyMethodDef cwr_methods[] = { {"__sizeof__", cwr_sizeof, METH_NOARGS, sizeof_doc}, {NULL, NULL} /* sentinel */ @@ -2908,7 +2953,7 @@ permutations_traverse(PyObject *op, visitproc visit, void *arg) } static PyObject * -permutations_next(PyObject *op) +permutations_next_lock_held(PyObject *op) { permutationsobject *po = permutationsobject_CAST(op); PyObject *elem; @@ -2998,6 +3043,16 @@ permutations_next(PyObject *op) return NULL; } +static PyObject * +permutations_next(PyObject *op) +{ + PyObject *result; + Py_BEGIN_CRITICAL_SECTION(op); + result = permutations_next_lock_held(op); + Py_END_CRITICAL_SECTION() + return result; +} + static PyMethodDef permuations_methods[] = { {"__sizeof__", permutations_sizeof, METH_NOARGS, sizeof_doc}, {NULL, NULL} /* sentinel */ @@ -3105,7 +3160,7 @@ accumulate_traverse(PyObject *op, visitproc visit, void *arg) } static PyObject * -accumulate_next(PyObject *op) +accumulate_next_lock_held(PyObject *op) { accumulateobject *lz = accumulateobject_CAST(op); PyObject *val, *newtotal; @@ -3137,6 +3192,16 @@ accumulate_next(PyObject *op) return newtotal; } +static PyObject * +accumulate_next(PyObject *op) +{ + PyObject *result; + Py_BEGIN_CRITICAL_SECTION(op); + result = accumulate_next_lock_held(op); + Py_END_CRITICAL_SECTION() + return result; +} + static PyType_Slot accumulate_slots[] = { {Py_tp_dealloc, accumulate_dealloc}, {Py_tp_getattro, PyObject_GenericGetAttr}, @@ -3895,7 +3960,7 @@ zip_longest_traverse(PyObject *op, visitproc visit, void *arg) } static PyObject * -zip_longest_next(PyObject *op) +zip_longest_next_lock_held(PyObject *op) { ziplongestobject *lz = ziplongestobject_CAST(op); Py_ssize_t i; @@ -3966,6 +4031,16 @@ zip_longest_next(PyObject *op) return result; } +static PyObject * +zip_longest_next(PyObject *op) +{ + PyObject *result; + Py_BEGIN_CRITICAL_SECTION(op); + result = zip_longest_next_lock_held(op); + Py_END_CRITICAL_SECTION() + return result; +} + PyDoc_STRVAR(zip_longest_doc, "zip_longest(*iterables, fillvalue=None)\n\ --\n\