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

https://github.com/python/cpython/commit/61818b6087e59de6287ac71bd734a0420d6e8ee3
commit: 61818b6087e59de6287ac71bd734a0420d6e8ee3
branch: main
author: Kumar Aditya <[email protected]>
committer: kumaraditya303 <[email protected]>
date: 2026-08-19T08:55:00+05:30
summary:

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

files:
A Lib/test/test_free_threading/test_async_generators.py
A Misc/NEWS.d/next/Core_and_Builtins/2026-08-01-18-04-11.gh-issue-120321.k3XvQb.rst
M Include/internal/pycore_interpframe_structs.h
M Lib/test/test_asyncgen.py
M Objects/genobject.c

diff --git a/Include/internal/pycore_interpframe_structs.h b/Include/internal/pycore_interpframe_structs.h
index 4d267e35504b94d..560fb66a97fc89d 100644
--- a/Include/internal/pycore_interpframe_structs.h
+++ b/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;                                    \
diff --git a/Lib/test/test_asyncgen.py b/Lib/test/test_asyncgen.py
index e9eecbc83415518..70a285dd91f385f 100644
--- a/Lib/test/test_asyncgen.py
+++ b/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():
diff --git a/Lib/test/test_free_threading/test_async_generators.py b/Lib/test/test_free_threading/test_async_generators.py
new file mode 100644
index 000000000000000..7987ff6ed2fafec
--- /dev/null
+++ b/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()
diff --git a/Misc/NEWS.d/next/Core_and_Builtins/2026-08-01-18-04-11.gh-issue-120321.k3XvQb.rst b/Misc/NEWS.d/next/Core_and_Builtins/2026-08-01-18-04-11.gh-issue-120321.k3XvQb.rst
new file mode 100644
index 000000000000000..e87dcb6f689985c
--- /dev/null
+++ b/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.
diff --git a/Objects/genobject.c b/Objects/genobject.c
index 3cdc06733363d3e..6529a66fc35a6b4 100644
--- a/Objects/genobject.c
+++ b/Objects/genobject.c
@@ -1604,6 +1604,44 @@ typedef enum {
     AWAITABLE_STATE_CLOSED, /* closed */
 } AwaitableState;
 
+#ifdef Py_GIL_DISABLED
+static bool
+async_gen_try_set_state(int8_t *state, int8_t *expected, int8_t new_state)
+{
+    return _Py_atomic_compare_exchange_int8(state, expected, new_state);
+}
+
+# define _Py_ASYNC_GEN_TRY_SET_STATE(state, expected, new_state) \
+    async_gen_try_set_state(&(state), &(expected), (new_state))
+#else
+# define _Py_ASYNC_GEN_TRY_SET_STATE(state, expected, new_state) \
+    ((state) = (new_state), true)
+#endif
+
+// Try to transition the async generator to the running state.
+// Returns false if it is already running.
+//
+// There are two ways to concurrently iterate an async generator: by
+// sharing a single asend()/athrow() object across threads, or with
+// multiple asend()/athrow() objects sending to the same generator.
+// The CAS on ags_state/agt_state handles the first case; the CAS on
+// ag_running_async here handles the second.
+static bool
+async_gen_try_claim_running(PyAsyncGenObject *agen)
+{
+#ifdef Py_GIL_DISABLED
+    int8_t expected = 0;
+    return _Py_atomic_compare_exchange_int8(&agen->ag_running_async,
+                                            &expected, 1);
+#else
+    if (agen->ag_running_async) {
+        return false;
+    }
+    agen->ag_running_async = 1;
+    return true;
+#endif
+}
+
 
 typedef struct PyAsyncGenASend {
     PyObject_HEAD
@@ -1613,7 +1651,7 @@ typedef struct PyAsyncGenASend {
        (equivalent of "asend(None)") */
     PyObject *ags_sendval;
 
-    AwaitableState ags_state;
+    int8_t ags_state;
 } PyAsyncGenASend;
 
 #define _PyAsyncGenASend_CAST(op) \
@@ -1630,7 +1668,7 @@ typedef struct PyAsyncGenAThrow {
     PyObject *agt_tb;
     PyObject *agt_val;
 
-    AwaitableState agt_state;
+    int8_t agt_state;
 } PyAsyncGenAThrow;
 
 
@@ -1672,11 +1710,17 @@ async_gen_init_hooks(PyAsyncGenObject *o)
     PyObject *finalizer;
     PyObject *firstiter;
 
+#ifdef Py_GIL_DISABLED
+    if (_Py_atomic_exchange_int8(&o->ag_hooks_inited, 1)) {
+        return 0;
+    }
+#else
     if (o->ag_hooks_inited) {
         return 0;
     }
 
     o->ag_hooks_inited = 1;
+#endif
 
     tstate = _PyThreadState_GET();
 
@@ -1919,10 +1963,9 @@ async_gen_unwrap_value(PyAsyncGenObject *gen, PyObject *result)
         if (PyErr_ExceptionMatches(PyExc_StopAsyncIteration)
             || PyErr_ExceptionMatches(PyExc_GeneratorExit)
         ) {
-            gen->ag_closed = 1;
+            FT_ATOMIC_STORE_INT8_RELAXED(gen->ag_closed, 1);
         }
 
-        gen->ag_running_async = 0;
         return NULL;
     }
 
@@ -1930,7 +1973,6 @@ async_gen_unwrap_value(PyAsyncGenObject *gen, PyObject *result)
         /* async yield */
         _PyGen_SetStopIterationValue(((_PyAsyncGenWrappedValue*)result)->agw_val);
         Py_DECREF(result);
-        gen->ag_running_async = 0;
         return NULL;
     }
 
@@ -1974,34 +2016,45 @@ static PyObject *
 async_gen_asend_send(PyObject *self, PyObject *arg)
 {
     PyAsyncGenASend *o = _PyAsyncGenASend_CAST(self);
-    if (o->ags_state == AWAITABLE_STATE_CLOSED) {
-        PyErr_SetString(
-            PyExc_RuntimeError,
-            "cannot reuse already awaited __anext__()/asend()");
-        return NULL;
-    }
 
-    if (o->ags_state == AWAITABLE_STATE_INIT) {
-        if (o->ags_gen->ag_running_async) {
-            o->ags_state = AWAITABLE_STATE_CLOSED;
+    int8_t state = FT_ATOMIC_LOAD_INT8_RELAXED(o->ags_state);
+    do {
+        if (state == AWAITABLE_STATE_CLOSED) {
             PyErr_SetString(
                 PyExc_RuntimeError,
-                "anext(): asynchronous generator is already running");
+                "cannot reuse already awaited __anext__()/asend()");
             return NULL;
         }
-
-        if (arg == NULL || arg == Py_None) {
-            arg = o->ags_sendval;
+        if (state == AWAITABLE_STATE_ITER) {
+            goto do_send;
         }
-        o->ags_state = AWAITABLE_STATE_ITER;
+        assert(state == AWAITABLE_STATE_INIT);
+    } while (!_Py_ASYNC_GEN_TRY_SET_STATE(o->ags_state, state,
+                                          AWAITABLE_STATE_ITER));
+
+    // The transition above only guards this object, the generator may
+    // still be running through another asend()/athrow() object so
+    // try to claim it before running.
+    if (!async_gen_try_claim_running(o->ags_gen)) {
+        FT_ATOMIC_STORE_INT8_RELAXED(o->ags_state, AWAITABLE_STATE_CLOSED);
+        PyErr_SetString(
+            PyExc_RuntimeError,
+            "anext(): asynchronous generator is already running");
+        return NULL;
     }
 
-    o->ags_gen->ag_running_async = 1;
-    PyObject *result = gen_send((PyObject*)o->ags_gen, arg);
+    if (arg == NULL || arg == Py_None) {
+        arg = o->ags_sendval;
+    }
+
+    PyObject *result;
+do_send:
+    result = gen_send((PyObject*)o->ags_gen, arg);
     result = async_gen_unwrap_value(o->ags_gen, result);
 
     if (result == NULL) {
-        o->ags_state = AWAITABLE_STATE_CLOSED;
+        FT_ATOMIC_STORE_INT8_RELAXED(o->ags_state, AWAITABLE_STATE_CLOSED);
+        FT_ATOMIC_STORE_INT8_RELEASE(o->ags_gen->ag_running_async, 0);
     }
 
     return result;
@@ -2033,32 +2086,40 @@ async_gen_asend_throw(PyObject *self, PyObject *const *args, Py_ssize_t nargs)
 {
     PyAsyncGenASend *o = _PyAsyncGenASend_CAST(self);
 
-    if (o->ags_state == AWAITABLE_STATE_CLOSED) {
-        PyErr_SetString(
-            PyExc_RuntimeError,
-            "cannot reuse already awaited __anext__()/asend()");
-        return NULL;
-    }
-
-    if (o->ags_state == AWAITABLE_STATE_INIT) {
-        if (o->ags_gen->ag_running_async) {
-            o->ags_state = AWAITABLE_STATE_CLOSED;
+    int8_t state = FT_ATOMIC_LOAD_INT8_RELAXED(o->ags_state);
+    do {
+        if (state == AWAITABLE_STATE_CLOSED) {
             PyErr_SetString(
                 PyExc_RuntimeError,
-                "anext(): asynchronous generator is already running");
+                "cannot reuse already awaited __anext__()/asend()");
             return NULL;
         }
-
-        o->ags_state = AWAITABLE_STATE_ITER;
-        o->ags_gen->ag_running_async = 1;
+        if (state == AWAITABLE_STATE_ITER) {
+            goto do_throw;
+        }
+        assert(state == AWAITABLE_STATE_INIT);
+    } while (!_Py_ASYNC_GEN_TRY_SET_STATE(o->ags_state, state,
+                                          AWAITABLE_STATE_ITER));
+
+    // The transition above only guards this object, the generator may
+    // still be running through another asend()/athrow() object so
+    // try to claim it before running.
+    if (!async_gen_try_claim_running(o->ags_gen)) {
+        FT_ATOMIC_STORE_INT8_RELAXED(o->ags_state, AWAITABLE_STATE_CLOSED);
+        PyErr_SetString(
+            PyExc_RuntimeError,
+            "anext(): asynchronous generator is already running");
+        return NULL;
     }
 
-    PyObject *result = gen_throw((PyObject*)o->ags_gen, args, nargs);
+    PyObject *result;
+do_throw:
+    result = gen_throw((PyObject*)o->ags_gen, args, nargs);
     result = async_gen_unwrap_value(o->ags_gen, result);
 
     if (result == NULL) {
-        o->ags_gen->ag_running_async = 0;
-        o->ags_state = AWAITABLE_STATE_CLOSED;
+        FT_ATOMIC_STORE_INT8_RELAXED(o->ags_state, AWAITABLE_STATE_CLOSED);
+        FT_ATOMIC_STORE_INT8_RELEASE(o->ags_gen->ag_running_async, 0);
     }
 
     return result;
@@ -2069,7 +2130,7 @@ static PyObject *
 async_gen_asend_close(PyObject *self, PyObject *args)
 {
     PyAsyncGenASend *o = _PyAsyncGenASend_CAST(self);
-    if (o->ags_state == AWAITABLE_STATE_CLOSED) {
+    if (FT_ATOMIC_LOAD_INT8_RELAXED(o->ags_state) == AWAITABLE_STATE_CLOSED) {
         Py_RETURN_NONE;
     }
 
@@ -2304,80 +2365,100 @@ async_gen_athrow_send(PyObject *self, PyObject *arg)
     PyGenObject *gen = _PyGen_CAST(o->agt_gen);
     PyObject *retval;
 
-    if (o->agt_state == AWAITABLE_STATE_CLOSED) {
+    int8_t state = FT_ATOMIC_LOAD_INT8_RELAXED(o->agt_state);
+    if (state == AWAITABLE_STATE_CLOSED) {
         PyErr_SetString(
             PyExc_RuntimeError,
             "cannot reuse already awaited aclose()/athrow()");
         return NULL;
     }
 
-    if (FRAME_STATE_FINISHED(gen->gi_frame_state)) {
-        o->agt_state = AWAITABLE_STATE_CLOSED;
+    if (FRAME_STATE_FINISHED(FT_ATOMIC_LOAD_INT8_RELAXED(gen->gi_frame_state))) {
+        // Close the awaitable, unless another thread transitioned it
+        // to a different state in the meantime.
+        (void)_Py_ASYNC_GEN_TRY_SET_STATE(o->agt_state, state,
+                                          AWAITABLE_STATE_CLOSED);
         PyErr_SetNone(PyExc_StopIteration);
         return NULL;
     }
 
-    if (o->agt_state == AWAITABLE_STATE_INIT) {
-        if (o->agt_gen->ag_running_async) {
-            o->agt_state = AWAITABLE_STATE_CLOSED;
-            if (o->agt_typ == NULL) {
-                PyErr_SetString(
-                    PyExc_RuntimeError,
-                    "aclose(): asynchronous generator is already running");
-            }
-            else {
-                PyErr_SetString(
-                    PyExc_RuntimeError,
-                    "athrow(): asynchronous generator is already running");
-            }
+    do {
+        if (state == AWAITABLE_STATE_CLOSED) {
+            PyErr_SetString(
+                PyExc_RuntimeError,
+                "cannot reuse already awaited aclose()/athrow()");
             return NULL;
         }
-
-        if (o->agt_gen->ag_closed) {
-            o->agt_state = AWAITABLE_STATE_CLOSED;
-            PyErr_SetNone(PyExc_StopAsyncIteration);
-            return NULL;
+        if (state == AWAITABLE_STATE_ITER) {
+            goto do_send;
         }
-
+        assert(state == AWAITABLE_STATE_INIT);
         if (arg != Py_None) {
             PyErr_SetString(PyExc_RuntimeError, NON_INIT_CORO_MSG);
             return NULL;
         }
+    } while (!_Py_ASYNC_GEN_TRY_SET_STATE(o->agt_state, state,
+                                          AWAITABLE_STATE_ITER));
+
+    // The transition above only guards this object, the generator may
+    // still be running through another asend()/athrow() object so
+    // try to claim it before running.
+    if (!async_gen_try_claim_running(o->agt_gen)) {
+        FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED);
+        if (o->agt_typ == NULL) {
+            PyErr_SetString(
+                PyExc_RuntimeError,
+                "aclose(): asynchronous generator is already running");
+        }
+        else {
+            PyErr_SetString(
+                PyExc_RuntimeError,
+                "athrow(): asynchronous generator is already running");
+        }
+        return NULL;
+    }
 
-        o->agt_state = AWAITABLE_STATE_ITER;
-        o->agt_gen->ag_running_async = 1;
+    if (FT_ATOMIC_LOAD_INT8_RELAXED(o->agt_gen->ag_closed)) {
+        FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED);
+        FT_ATOMIC_STORE_INT8_RELEASE(o->agt_gen->ag_running_async, 0);
+        PyErr_SetNone(PyExc_StopAsyncIteration);
+        return NULL;
+    }
 
-        if (o->agt_typ == NULL) {
-            /* aclose() mode */
-            o->agt_gen->ag_closed = 1;
+    if (o->agt_typ == NULL) {
+        /* aclose() mode */
+        FT_ATOMIC_STORE_INT8_RELAXED(o->agt_gen->ag_closed, 1);
 
-            retval = _gen_throw((PyGenObject *)gen,
-                                0,  /* Do not close generator when
-                                       PyExc_GeneratorExit is passed */
-                                PyExc_GeneratorExit, NULL, NULL);
+        retval = _gen_throw((PyGenObject *)gen,
+                            0,  /* Do not close generator when
+                                   PyExc_GeneratorExit is passed */
+                            PyExc_GeneratorExit, NULL, NULL);
 
-            if (retval && _PyAsyncGenWrappedValue_CheckExact(retval)) {
-                Py_DECREF(retval);
-                goto yield_close;
-            }
-        } else {
-            retval = _gen_throw((PyGenObject *)gen,
-                                0,  /* Do not close generator when
-                                       PyExc_GeneratorExit is passed */
-                                o->agt_typ, o->agt_val, o->agt_tb);
-            retval = async_gen_unwrap_value(o->agt_gen, retval);
-        }
-        if (retval == NULL) {
-            goto check_error;
+        if (retval && _PyAsyncGenWrappedValue_CheckExact(retval)) {
+            Py_DECREF(retval);
+            goto yield_close;
         }
-        return retval;
+    } else {
+        retval = _gen_throw((PyGenObject *)gen,
+                            0,  /* Do not close generator when
+                                   PyExc_GeneratorExit is passed */
+                            o->agt_typ, o->agt_val, o->agt_tb);
+        retval = async_gen_unwrap_value(o->agt_gen, retval);
     }
+    if (retval == NULL) {
+        goto check_error;
+    }
+    return retval;
 
-    assert(o->agt_state == AWAITABLE_STATE_ITER);
-
+do_send:
     retval = gen_send((PyObject *)gen, arg);
     if (o->agt_typ) {
-        return async_gen_unwrap_value(o->agt_gen, retval);
+        retval = async_gen_unwrap_value(o->agt_gen, retval);
+        if (retval == NULL) {
+            FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED);
+            FT_ATOMIC_STORE_INT8_RELEASE(o->agt_gen->ag_running_async, 0);
+        }
+        return retval;
     } else {
         /* aclose() mode */
         if (retval) {
@@ -2395,15 +2476,15 @@ async_gen_athrow_send(PyObject *self, PyObject *arg)
     }
 
 yield_close:
-    o->agt_gen->ag_running_async = 0;
-    o->agt_state = AWAITABLE_STATE_CLOSED;
+    FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED);
+    FT_ATOMIC_STORE_INT8_RELEASE(o->agt_gen->ag_running_async, 0);
     PyErr_SetString(
         PyExc_RuntimeError, ASYNC_GEN_IGNORED_EXIT_MSG);
     return NULL;
 
 check_error:
-    o->agt_gen->ag_running_async = 0;
-    o->agt_state = AWAITABLE_STATE_CLOSED;
+    FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED);
+    FT_ATOMIC_STORE_INT8_RELEASE(o->agt_gen->ag_running_async, 0);
     if (PyErr_ExceptionMatches(PyExc_StopAsyncIteration) ||
             PyErr_ExceptionMatches(PyExc_GeneratorExit))
     {
@@ -2426,54 +2507,62 @@ async_gen_athrow_throw(PyObject *self, PyObject *const *args, Py_ssize_t nargs)
 {
     PyAsyncGenAThrow *o = _PyAsyncGenAThrow_CAST(self);
 
-    if (o->agt_state == AWAITABLE_STATE_CLOSED) {
-        PyErr_SetString(
-            PyExc_RuntimeError,
-            "cannot reuse already awaited aclose()/athrow()");
-        return NULL;
-    }
-
-    if (o->agt_state == AWAITABLE_STATE_INIT) {
-        if (o->agt_gen->ag_running_async) {
-            o->agt_state = AWAITABLE_STATE_CLOSED;
-            if (o->agt_typ == NULL) {
-                PyErr_SetString(
-                    PyExc_RuntimeError,
-                    "aclose(): asynchronous generator is already running");
-            }
-            else {
-                PyErr_SetString(
-                    PyExc_RuntimeError,
-                    "athrow(): asynchronous generator is already running");
-            }
+    int8_t state = FT_ATOMIC_LOAD_INT8_RELAXED(o->agt_state);
+    do {
+        if (state == AWAITABLE_STATE_CLOSED) {
+            PyErr_SetString(
+                PyExc_RuntimeError,
+                "cannot reuse already awaited aclose()/athrow()");
             return NULL;
         }
-
-        o->agt_state = AWAITABLE_STATE_ITER;
-        o->agt_gen->ag_running_async = 1;
+        if (state == AWAITABLE_STATE_ITER) {
+            goto do_throw;
+        }
+        assert(state == AWAITABLE_STATE_INIT);
+    } while (!_Py_ASYNC_GEN_TRY_SET_STATE(o->agt_state, state,
+                                          AWAITABLE_STATE_ITER));
+
+    // The transition above only guards this object, the generator may
+    // still be running through another asend()/athrow() object so
+    // try to claim it before running.
+    if (!async_gen_try_claim_running(o->agt_gen)) {
+        FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED);
+        if (o->agt_typ == NULL) {
+            PyErr_SetString(
+                PyExc_RuntimeError,
+                "aclose(): asynchronous generator is already running");
+        }
+        else {
+            PyErr_SetString(
+                PyExc_RuntimeError,
+                "athrow(): asynchronous generator is already running");
+        }
+        return NULL;
     }
 
-    PyObject *retval = gen_throw((PyObject*)o->agt_gen, args, nargs);
+    PyObject *retval;
+do_throw:
+    retval = gen_throw((PyObject*)o->agt_gen, args, nargs);
     if (o->agt_typ) {
         retval = async_gen_unwrap_value(o->agt_gen, retval);
         if (retval == NULL) {
-            o->agt_gen->ag_running_async = 0;
-            o->agt_state = AWAITABLE_STATE_CLOSED;
+            FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED);
+            FT_ATOMIC_STORE_INT8_RELEASE(o->agt_gen->ag_running_async, 0);
         }
         return retval;
     }
     else {
         /* aclose() mode */
         if (retval && _PyAsyncGenWrappedValue_CheckExact(retval)) {
-            o->agt_gen->ag_running_async = 0;
-            o->agt_state = AWAITABLE_STATE_CLOSED;
+            FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED);
+            FT_ATOMIC_STORE_INT8_RELEASE(o->agt_gen->ag_running_async, 0);
             Py_DECREF(retval);
             PyErr_SetString(PyExc_RuntimeError, ASYNC_GEN_IGNORED_EXIT_MSG);
             return NULL;
         }
         if (retval == NULL) {
-            o->agt_gen->ag_running_async = 0;
-            o->agt_state = AWAITABLE_STATE_CLOSED;
+            FT_ATOMIC_STORE_INT8_RELAXED(o->agt_state, AWAITABLE_STATE_CLOSED);
+            FT_ATOMIC_STORE_INT8_RELEASE(o->agt_gen->ag_running_async, 0);
         }
         if (PyErr_ExceptionMatches(PyExc_StopAsyncIteration) ||
             PyErr_ExceptionMatches(PyExc_GeneratorExit))
@@ -2502,7 +2591,7 @@ static PyObject *
 async_gen_athrow_close(PyObject *self, PyObject *args)
 {
     PyAsyncGenAThrow *agt = _PyAsyncGenAThrow_CAST(self);
-    if (agt->agt_state == AWAITABLE_STATE_CLOSED) {
+    if (FT_ATOMIC_LOAD_INT8_RELAXED(agt->agt_state) == AWAITABLE_STATE_CLOSED) {
         Py_RETURN_NONE;
     }
     PyObject *result = async_gen_athrow_throw((PyObject*)agt,

_______________________________________________
Python-checkins mailing list -- [email protected]
To unsubscribe send an email to [email protected]
https://mail.python.org/mailman3//lists/python-checkins.python.org
Member address: [email protected]