gh-160347: Fix concurrent driving race on async generator awaitables

Lib/test/test_asyncgen.py

@@ -665,6 +665,84 @@ async def agenfn():
             gen.send(None)
         self.assertEqual(cm.exception.value, 2)
 
+    def test_async_gen_awaitable_reentrant_send(self):
+        class YieldOnce:
+            def __await__(self):
+                yield "inner"
+
+        aw = None
+        seen_errors = []
+
+        async def reenter_on_asend():
+            try:
+                aw.send(None)
+            except ValueError as exc:
+                seen_errors.append(str(exc))
+            await YieldOnce()
+            yield 42
+
+        ag = reenter_on_asend()
+        aw = ag.asend(None)
+        self.assertEqual(aw.send(None), "inner")
+        self.assertEqual(seen_errors, ["async generator already executing"])
+        self.assertTrue(ag.ag_running)
+        with self.assertRaises(StopIteration) as cm:
+            aw.send(None)
+        self.assertEqual(cm.exception.value, 42)
+        self.assertFalse(ag.ag_running)
+        with self.assertRaisesRegex(RuntimeError, "cannot reuse"):
+            aw.send(None)
+
+        seen_errors.clear()
+
+        async def reenter_on_aclose():
+            try:
+                yield 1
+            finally:
+                try:
+                    aw.send(None)
+                except ValueError as exc:
+                    seen_errors.append(("send", str(exc)))
+                try:
+                    aw.throw(RuntimeError("boom"))
+                except ValueError as exc:
+                    seen_errors.append(("throw", str(exc)))
+                await YieldOnce()
+
+        ag = reenter_on_aclose()
+        with self.assertRaises(StopIteration):
+            ag.asend(None).send(None)
+        aw = ag.aclose()
+        self.assertEqual(aw.send(None), "inner")
+        self.assertEqual(seen_errors, [
+            ("send", "async generator already executing"),
+            ("throw", "async generator already executing"),
+        ])
+        self.assertTrue(ag.ag_running)
+        with self.assertRaises(StopIteration):
+            aw.send(None)
+        self.assertFalse(ag.ag_running)
+        with self.assertRaisesRegex(RuntimeError, "cannot reuse"):
+            aw.send(None)
+
+    def test_async_gen_finished_aclose_athrow_non_none_send(self):
+        async def empty():
+            if False:
+                yield
+
+        for make_aw in (
+            lambda ag: ag.aclose(),
+            lambda ag: ag.athrow(ValueError),
+        ):
+            ag = empty()
+            with self.assertRaises(StopAsyncIteration):
+                ag.asend(None).send(None)
+            aw = make_aw(ag)
+            with self.assertRaises(StopIteration):
+                aw.send(1)
+            with self.assertRaises(RuntimeError):
+                aw.send(None)
+
 
 class AsyncGenAsyncioTest(unittest.TestCase):
 

Lib/test/test_free_threading/test_async_generators.py

@@ -1,6 +1,8 @@
 import sys
+import threading
 import unittest
 
+from test import support
 from test.support import threading_helper
 
 threading_helper.requires_working_threading(module=True)
@@ -243,6 +245,170 @@ def worker():
         threading_helper.run_concurrently(worker, self.NUM_THREADS)
         self.assertEqual(calls, [ag])
 
+    def test_concurrent_shared_aclose_does_not_resume(self):
+        # gh-160347: concurrent calls on a shared aclose() awaitable must
+        # never resume the async generator past its suspended yield.
+        rounds = 500
+        start_barrier = threading.Barrier(self.NUM_THREADS + 1)
+        done_barrier = threading.Barrier(self.NUM_THREADS + 1)
+        state = {"aw": None, "stop": False}
+        errors = []
+
+        def worker():
+            try:
+                while True:
+                    start_barrier.wait(timeout=support.SHORT_TIMEOUT)
+                    if state["stop"]:
+                        return
+                    try:
+                        state["aw"].send(None)
+                    except (RuntimeError, ValueError,
+                            StopIteration, StopAsyncIteration):
+                        pass
+                    done_barrier.wait(timeout=support.SHORT_TIMEOUT)
+            except threading.BrokenBarrierError:
+                return
+            except BaseException as exc:
+                errors.append(exc)
+                start_barrier.abort()
+                done_barrier.abort()
+
+        threads = [threading.Thread(target=worker)
+                   for _ in range(self.NUM_THREADS)]
+        started = []
+        try:
+            for t in threads:
+                t.start()
+                started.append(t)
+
+            for _ in range(rounds):
+                cleanups = []
+                resumed = []
+
+                async def agen():
+                    try:
+                        yield 1
+                    finally:
+                        cleanups.append(1)
+                    resumed.append(1)
+                    yield 2
+
+                ag = agen()
+                with self.assertRaises(StopIteration):
+                    ag.asend(None).send(None)
+
+                aw = ag.aclose()
+                state["aw"] = aw
+                start_barrier.wait(timeout=support.SHORT_TIMEOUT)
+                done_barrier.wait(timeout=support.SHORT_TIMEOUT)
+
+                self.assertEqual(cleanups, [1])
+                self.assertEqual(resumed, [])
+                self.assertFalse(ag.ag_running)
+                self.assertRaises(RuntimeError, aw.send, None)
+                self.assertRaises(
+                    StopAsyncIteration, ag.asend(None).send, None)
+        finally:
+            state["stop"] = True
+            try:
+                if not errors and len(started) == len(threads):
+                    start_barrier.wait(timeout=support.SHORT_TIMEOUT)
+                else:
+                    start_barrier.abort()
+                    done_barrier.abort()
+            except threading.BrokenBarrierError:
+                start_barrier.abort()
+                done_barrier.abort()
+            for t in started:
+                t.join(timeout=support.SHORT_TIMEOUT)
+                self.assertFalse(t.is_alive())
+            self.assertEqual(errors, [])
+
+    def test_concurrent_shared_suspended_awaitable(self):
+        # Concurrent calls on a shared awaitable that is suspended at
+        # AWAITABLE_STATE_ITER must fail the racing caller without corrupting
+        # ag_running or closing the awaitable while the active step finishes.
+        for kind in ("asend", "athrow", "aclose"):
+            driver_in_step = threading.Event()
+            racer_done = threading.Event()
+            racer_errors = []
+
+            class SyncYieldTwice:
+                def __await__(self):
+                    yield "first"
+                    driver_in_step.set()
+                    self.wait_ok = racer_done.wait(
+                        timeout=support.SHORT_TIMEOUT)
+                    yield "second"
+
+            sync = SyncYieldTwice()
+
+            async def agen_asend():
+                await sync
+                yield 2
+
+            async def agen_athrow():
+                try:
+                    yield 1
+                except ValueError:
+                    await sync
+                    yield 2
+
+            async def agen_aclose():
+                try:
+                    yield 1
+                except GeneratorExit:
+                    await sync
+                    raise
+
+            if kind == "asend":
+                ag = agen_asend()
+                aw = ag.asend(None)
+            elif kind == "athrow":
+                ag = agen_athrow()
+                with self.assertRaises(StopIteration):
+                    ag.asend(None).send(None)
+                aw = ag.athrow(ValueError)
+            else:
+                ag = agen_aclose()
+                with self.assertRaises(StopIteration):
+                    ag.asend(None).send(None)
+                aw = ag.aclose()
+
+            # Advance to "first" so aw is suspended in AWAITABLE_STATE_ITER.
+            self.assertEqual(aw.send(None), "first")
+            self.assertTrue(ag.ag_running)
+
+            driver_results = []
+
+            def driver():
+                driver_results.append(aw.send(None))
+
+            def racer():
+                if not driver_in_step.wait(timeout=support.SHORT_TIMEOUT):
+                    racer_errors.append(
+                        TimeoutError("driver did not enter step"))
+                    return
+                try:
+                    aw.send(None)
+                except ValueError as exc:
+                    racer_errors.append(exc)
+                finally:
+                    racer_done.set()
+
+            threading_helper.run_concurrently([driver, racer])
+            self.assertTrue(sync.wait_ok)
+            self.assertEqual(driver_results, ["second"])
+            self.assertEqual(len(racer_errors), 1)
+            self.assertIsInstance(racer_errors[0], ValueError)
+            self.assertIn("already executing", str(racer_errors[0]))
+            self.assertTrue(ag.ag_running)
+
+            with self.assertRaises(StopIteration):
+                aw.send(None)
+            self.assertFalse(ag.ag_running)
+            self.assertRaises(RuntimeError, aw.send, None)
+
 
 if __name__ == "__main__":
     unittest.main()

Misc/NEWS.d/next/Core_and_Builtins/2026-11-14-11-49-16.gh-issue-160347.JirvQ7.rst

@@ -0,0 +1,4 @@
+Fix state corruption caused by concurrent or reentrant calls on the same async
+generator awaitable (``asend()``, ``athrow()``, or ``aclose()``). This could
+prevent an ``aclose()`` operation from closing the generator or make an active
+awaitable unusable.