gh-120321: fix thread safety of concurrently iterating over async generators (#155025)

Author: kumaraditya303

Include/internal/pycore_interpframe_structs.h

@@ -66,9 +66,9 @@ struct _PyInterpreterFrame {
     PyObject *prefix##_qualname;                                            \
     _PyErr_StackItem prefix##_exc_state;                                    \
     PyObject *prefix##_origin_or_finalizer;                                 \
-    char prefix##_hooks_inited;                                             \
-    char prefix##_closed;                                                   \
-    char prefix##_running_async;                                            \
+    int8_t prefix##_hooks_inited;                                           \
+    int8_t prefix##_closed;                                                 \
+    int8_t prefix##_running_async;                                          \
     /* The frame */                                                         \
     int8_t prefix##_frame_state;                                            \
     _PyInterpreterFrame prefix##_iframe;                                    \

Lib/test/test_asyncgen.py

@@ -619,6 +619,33 @@ async def agenfn():
         with self.assertRaisesRegex(RuntimeError, "coroutine ignored GeneratorExit"):
             gen.close()
 
+    def test_async_gen_athrow_send_non_none(self):
+        # gh-120321: sending a non-None value to a just-started athrow()
+        # awaitable must not claim the generator, so the generator stays
+        # usable and the awaitable can still be awaited afterwards.
+        class MyExc(Exception):
+            pass
+
+        async def agenfn():
+            try:
+                yield 1
+            except MyExc:
+                yield 2
+
+        agen = agenfn()
+        with self.assertRaises(StopIteration):
+            agen.asend(None).send(None)
+
+        gen = agen.athrow(MyExc)
+        with self.assertRaisesRegex(RuntimeError, "non-None value"):
+            gen.send(42)
+        self.assertFalse(agen.ag_running)
+
+        # The awaitable is still in its initial state and works normally.
+        with self.assertRaises(StopIteration) as cm:
+            gen.send(None)
+        self.assertEqual(cm.exception.value, 2)
+
 
 class AsyncGenAsyncioTest(unittest.TestCase):
 
@@ -1950,6 +1977,41 @@ class MyException(Exception):
         ):
             nxt.throw(MyException)
 
+    def test_async_gen_send_same_athrow_coro_after_completion(self):
+        # gh-120321: an athrow() awaitable that needs more than one send()
+        # to complete must be closed on completion; sending to it again
+        # must raise instead of resuming the generator.
+        class YieldOnce:
+            def __await__(self):
+                yield
+
+        async def async_iterate():
+            try:
+                yield 1
+            except ValueError:
+                await YieldOnce()
+            yield 2
+
+        it = async_iterate()
+        with self.assertRaises(StopIteration):
+            it.__anext__().send(None)
+
+        nxt = it.athrow(ValueError)
+        # The exception handler suspends before the operation completes.
+        nxt.send(None)
+        with self.assertRaises(StopIteration) as cm:
+            nxt.send(None)
+        self.assertEqual(cm.exception.value, 2)
+
+        with self.assertRaisesRegex(
+            RuntimeError,
+            r"cannot reuse already awaited aclose\(\)/athrow\(\)"
+        ):
+            nxt.send(None)
+
+        with self.assertRaises(StopIteration):
+            it.aclose().send(None)
+
     def test_async_gen_aclose_twice_with_different_coros(self):
         # Regression test for https://bugs.python.org/issue39606
         async def async_iterate():

Lib/test/test_free_threading/test_async_generators.py

@@ -0,0 +1,248 @@
+import sys
+import unittest
+
+from test.support import threading_helper
+
+threading_helper.requires_working_threading(module=True)
+
+
+class TestFTAsyncGenerators(unittest.TestCase):
+    NUM_THREADS = 4
+
+    def test_concurrent_anext(self):
+        # Each yielded value must be delivered to exactly one thread.
+        async def agen():
+            for i in range(100):
+                yield i
+
+        ag = agen()
+        values = []
+
+        def drive():
+            while True:
+                try:
+                    ag.asend(None).send(None)
+                except StopIteration as e:
+                    values.append(e.value)
+                except StopAsyncIteration:
+                    break
+                except RuntimeError:
+                    # Another thread is currently driving the generator.
+                    continue
+
+        threading_helper.run_concurrently(drive, self.NUM_THREADS)
+        self.assertEqual(sorted(values), list(range(100)))
+
+    def test_concurrent_athrow(self):
+        # Each thrown exception must be delivered to the generator
+        # exactly once.
+        received = []
+
+        async def agen():
+            while True:
+                try:
+                    yield 1
+                except ValueError:
+                    received.append(1)
+
+        ag = agen()
+        with self.assertRaises(StopIteration):
+            ag.asend(None).send(None)  # advance to the first yield
+
+        delivered = []
+
+        def worker():
+            for _ in range(50):
+                try:
+                    ag.athrow(ValueError).send(None)
+                except StopIteration as e:
+                    # The generator received the exception and yielded again.
+                    delivered.append(e.value)
+                except RuntimeError:
+                    # Another thread is currently driving the generator.
+                    pass
+
+        threading_helper.run_concurrently(worker, self.NUM_THREADS)
+        self.assertEqual(len(received), len(delivered))
+
+    def test_concurrent_aclose(self):
+        # The generator must be cleaned up exactly once.
+        cleanups = []
+
+        async def agen():
+            try:
+                while True:
+                    yield 1
+            finally:
+                cleanups.append(1)
+
+        ag = agen()
+        with self.assertRaises(StopIteration):
+            ag.asend(None).send(None)  # advance to the first yield
+
+        def worker():
+            try:
+                ag.aclose().send(None)
+            except StopIteration:
+                # aclose() completed.
+                pass
+            except StopAsyncIteration:
+                # The generator was already closed.
+                pass
+            except RuntimeError:
+                # Another thread is currently driving the generator.
+                pass
+
+        threading_helper.run_concurrently(worker, self.NUM_THREADS)
+        self.assertEqual(cleanups, [1])
+        self.assertRaises(StopAsyncIteration, ag.asend(None).send, None)
+
+    def test_concurrent_shared_asend(self):
+        # Multiple threads racing on a single asend awaitable: the value
+        # must be delivered exactly once.
+        async def agen():
+            yield 1
+
+        ag = agen()
+        aw = ag.asend(None)
+        results = []
+
+        def worker():
+            try:
+                aw.send(None)
+            except StopIteration as e:
+                results.append(e.value)
+            except (RuntimeError, ValueError, StopAsyncIteration):
+                pass
+
+        threading_helper.run_concurrently(worker, self.NUM_THREADS)
+        self.assertEqual(results, [1])
+        # The awaitable is closed after the operation completed.
+        self.assertRaises(RuntimeError, aw.send, None)
+
+    def test_concurrent_shared_athrow(self):
+        # Multiple threads racing on a single athrow awaitable.
+        async def agen():
+            while True:
+                yield 1
+
+        ag = agen()
+        with self.assertRaises(StopIteration):
+            ag.asend(None).send(None)  # advance to the first yield
+        aw = ag.athrow(ValueError)
+
+        def worker():
+            try:
+                aw.send(None)
+            except (RuntimeError, ValueError,
+                    StopIteration, StopAsyncIteration):
+                pass
+
+        threading_helper.run_concurrently(worker, self.NUM_THREADS)
+        self.assertRaises(StopAsyncIteration, ag.asend(None).send, None)
+        # The awaitable is closed after the operation completed.
+        self.assertRaises(RuntimeError, aw.send, None)
+
+    def test_concurrent_shared_aclose(self):
+        # Multiple threads racing on a single aclose awaitable: the
+        # generator must be cleaned up exactly once.
+        cleanups = []
+
+        async def agen():
+            try:
+                while True:
+                    yield 1
+            finally:
+                cleanups.append(1)
+
+        ag = agen()
+        with self.assertRaises(StopIteration):
+            ag.asend(None).send(None)  # advance to the first yield
+        aw = ag.aclose()
+
+        def worker():
+            try:
+                aw.send(None)
+            except (RuntimeError, ValueError,
+                    StopIteration, StopAsyncIteration):
+                pass
+
+        threading_helper.run_concurrently(worker, self.NUM_THREADS)
+        self.assertEqual(cleanups, [1])
+        self.assertRaises(StopAsyncIteration, ag.asend(None).send, None)
+        # The awaitable is closed after the operation completed.
+        self.assertRaises(RuntimeError, aw.send, None)
+
+    def test_concurrent_anext_athrow(self):
+        async def agen():
+            while True:
+                try:
+                    yield 1
+                except ValueError:
+                    pass
+
+        ag = agen()
+
+        def worker():
+            for i in range(100):
+                try:
+                    if i % 2:
+                        ag.asend(None).send(None)
+                    else:
+                        ag.athrow(ValueError).send(None)
+                except (RuntimeError, ValueError,
+                        StopIteration, StopAsyncIteration):
+                    pass
+
+        threading_helper.run_concurrently(worker, self.NUM_THREADS)
+
+    def test_concurrent_anext_aclose(self):
+        async def agen():
+            for i in range(100):
+                yield i
+
+        ag = agen()
+
+        def anext_worker():
+            for _ in range(100):
+                try:
+                    ag.asend(None).send(None)
+                except (RuntimeError, ValueError,
+                        StopIteration, StopAsyncIteration):
+                    pass
+
+        def aclose_worker():
+            for _ in range(100):
+                try:
+                    ag.aclose().send(None)
+                except (RuntimeError, ValueError,
+                        StopIteration, StopAsyncIteration):
+                    pass
+
+        threading_helper.run_concurrently(
+            [anext_worker, aclose_worker, anext_worker, aclose_worker])
+
+    def test_firstiter_hook_called_once(self):
+        # Racing the first iteration must invoke the firstiter hook
+        # exactly once.
+        async def agen():
+            yield 1
+
+        ag = agen()
+        calls = []
+
+        def worker():
+            # Async generator hooks are per-thread state.
+            sys.set_asyncgen_hooks(firstiter=calls.append)
+            try:
+                ag.asend(None).send(None)
+            except (RuntimeError, ValueError,
+                    StopIteration, StopAsyncIteration):
+                pass
+
+        threading_helper.run_concurrently(worker, self.NUM_THREADS)
+        self.assertEqual(calls, [ag])
+
+
+if __name__ == "__main__":
+    unittest.main()

Misc/NEWS.d/next/Core_and_Builtins/2026-08-01-18-04-11.gh-issue-120321.k3XvQb.rst

@@ -0,0 +1,5 @@
+Fix thread safety of :term:`asynchronous generators <asynchronous
+generator>` when iterated, closed or thrown into concurrently from multiple
+threads on the :term:`free threading` build. Also make an ``athrow()``
+awaitable single-use: reusing it after completion now raises
+:exc:`RuntimeError` instead of resuming the generator again.