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]
