gh-124397: Add free-threading support for iterators.

Doc/library/threading.rst

@@ -1436,3 +1436,159 @@ is equivalent to::
 Currently, :class:`Lock`, :class:`RLock`, :class:`Condition`,
 :class:`Semaphore`, and :class:`BoundedSemaphore` objects may be used as
 :keyword:`with` statement context managers.
+
+
+Iterator synchronization
+------------------------
+
+By default, Python iterators do not support concurrent access. Most iterators make
+no guarantees when accessed simultaneously from multiple threads. Generator
+iterators, for example, raise :exc:`ValueError` if one of their iterator methods
+is called while the generator is already executing. The tools in this section
+allow reliable concurrency support to be added to ordinary iterators and
+iterator-producing callables.
+
+The :class:`serialize_iterator` wrapper lets multiple threads share a single iterator and
+take turns consuming from it. While one thread is running ``__next__()``, the
+others block until the iterator becomes available. Each value produced by the
+underlying iterator is delivered to exactly one caller.
+
+The :func:`concurrent_tee` function lets multiple threads each receive the full
+stream of values from one underlying iterator. It creates independent iterators
+that all draw from the same source. Values are buffered until consumed by all
+of the derived iterators.
+
+.. class:: serialize_iterator(iterable)
+
+ Return an iterator wrapper that serializes concurrent calls to
+ :meth:`~iterator.__next__` using a lock.
+
+ If the wrapped iterator also defines :meth:`~generator.send`,
+ :meth:`~generator.throw`, or :meth:`~generator.close`, those calls
+ are serialized as well.
+
+ This makes it possible to share a single iterator, including a generator
+ iterator, between multiple threads. A lock ensures that calls are handled
+ one at a time. No values are duplicated or skipped by the wrapper itself.
+ Each item from the underlying iterator is given to exactly one caller.
+
+ This wrapper does not copy or buffer values. Threads that call
+ :func:`next` while another thread is already advancing the iterator will
+ block until the active call completes.
+
+ Example:
+
+ .. code-block:: python
+
+ import threading
+
+ def squares(n):
+ for x in range(n):
+ yield x * x
+
+ def consume(name, iterable):
+ for item in iterable:
+ print(name, item)
+
+ source = threading.serialize_iterator(squares(5))
+
+ t1 = threading.Thread(target=consume, args=("left", source))
+ t2 = threading.Thread(target=consume, args=("right", source))
+ t1.start()
+ t2.start()
+ t1.join()
+ t2.join()
+
+ In this example, each number is printed exactly once, but the work is shared
+ between the two threads.
+
+ .. versionadded:: next
+
+
+.. function:: synchronized_iterator(func)
+
+ Wrap an iterator-producing callable so that each iterator it returns is
+ automatically passed through :class:`serialize_iterator`.
+
+ This is especially useful as a :term:`decorator` for generator functions,
+ allowing their generator-iterators to be consumed from multiple threads.
+
+ Example:
+
+ .. code-block:: python
+
+ import threading
+
+ @threading.synchronized_iterator
+ def squares(n):
+ for x in range(n):
+ yield x * x
+
+ def consume(name, iterable):
+ for item in iterable:
+ print(name, item)
+
+ source = squares(5)
+
+ t1 = threading.Thread(target=consume, args=("left", source))
+ t2 = threading.Thread(target=consume, args=("right", source))
+ t1.start()
+ t2.start()
+ t1.join()
+ t2.join()
+
+ The returned wrapper preserves the metadata of *func*, such as its name and
+ wrapped function reference.
+
+ .. versionadded:: next
+
+
+.. function:: concurrent_tee(iterable, n=2)
+
+ Return *n* independent iterators from a single input *iterable*, with
+ guaranteed behavior when the derived iterators are consumed concurrently.
+
+ This function is similar to :func:`itertools.tee`, but is intended for cases
+ where the source iterator may feed consumers running in different threads.
+ Each returned iterator yields every value from the underlying iterable, in
+ the same order.
+
+ Internally, values are buffered until every derived iterator has consumed
+ them.
+
+ The returned iterators share the same underlying synchronization lock. Each
+ individual derived iterator is intended to be consumed by one thread at a
+ time. If a single derived iterator must itself be shared by multiple
+ threads, wrap it with :class:`serialize_iterator`.
+
+ If *n* is ``0``, return an empty tuple. If *n* is negative, raise
+ :exc:`ValueError`.
+
+ Example:
+
+ .. code-block:: python
+
+ import threading
+
+ def squares(n):
+ for x in range(n):
+ yield x * x
+
+ def consume(name, iterable):
+ for item in iterable:
+ print(name, item)
+
+ source = squares(5)
+ left, right = threading.concurrent_tee(source)
+
+ t1 = threading.Thread(target=consume, args=("left", left))
+ t2 = threading.Thread(target=consume, args=("right", right))
+ t1.start()
+ t2.start()
+ t1.join()
+ t2.join()
+
+ In this example, both consumer threads see the full sequence of squares
+ from a single generator expression.
+
+ .. versionadded:: next

Doc/whatsnew/3.15.rst

@@ -1278,6 +1278,16 @@ tarfile
  (Contributed by Christoph Walcher in :gh:`57911`.)
 
 
+threading
+---------
+
+* Added :class:`~threading.serialize_iterator`,
+ :func:`~threading.synchronized_iterator`,
+ and :func:`~threading.concurrent_tee` to support concurrent access to
+ generators and iterators.
+ (Contributed by Raymond Hettinger in :gh:`124397`.)
+
+
 timeit
 ------