Skip to content

Commit ba728ed

Browse files
committed
gh-157044: Preserve asyncio call stacks through callable aiter()
1 parent 23180c5 commit ba728ed

6 files changed

Lines changed: 158 additions & 1 deletion

File tree

Doc/library/functions.rst

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -87,6 +87,13 @@ are always available. They are listed here in alphabetical order.
8787
The callable is only called when the result of :meth:`~object.__anext__`
8888
is awaited.
8989

90+
.. impl-detail::
91+
92+
The awaitable returned by :meth:`~object.__anext__` exposes the object
93+
returned by *callable* through its read-only ``aw_wrapped`` attribute.
94+
This attribute is ``None`` until the callable is invoked, and retains
95+
the returned object after the awaitable completes or is closed.
96+
9097
*stop_exception* is an exception class or a tuple of exception classes.
9198
If *stop_value* is not specified,
9299
the iteration stops only when the callable raises an exception.

Lib/asyncio/graph.py

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -65,6 +65,9 @@ def _build_graph_for_future(
6565
# A native async generator or duck-type compatible iterator
6666
st.append(FrameCallGraphEntry(coro.ag_frame))
6767
coro = coro.ag_await
68+
elif hasattr(coro, 'aw_wrapped'):
69+
# An asynchronous callable iterator's frameless awaitable.
70+
coro = coro.aw_wrapped
6871
else:
6972
break
7073

Lib/test/test_asyncgen.py

Lines changed: 81 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -909,12 +909,87 @@ async def spam():
909909
def test_aiter_callable_awaitable(self):
910910
it = aiter(self.make_counter(), 10)
911911
awaitable = it.__anext__()
912+
self.assertIsNone(awaitable.aw_wrapped)
912913
self.assertIsNone(awaitable.close())
914+
self.assertIsNone(awaitable.aw_wrapped)
913915
with self.assertRaises(RuntimeError):
914916
self.loop.run_until_complete(awaitable)
915917
awaitable = it.__anext__()
916918
with self.assertRaises(KeyError):
917919
awaitable.throw(KeyError('injected'))
920+
self.assertIsNone(awaitable.aw_wrapped)
921+
922+
def test_aiter_callable_wrapped(self):
923+
async def produce():
924+
await awaitable()
925+
await awaitable()
926+
return 1
927+
928+
class CustomAwaitable:
929+
def __init__(self):
930+
self.iterator = self.iterate()
931+
932+
def iterate(self):
933+
yield ('result',)
934+
935+
def __await__(self):
936+
return self.iterator
937+
938+
for factory in (produce, awaitable, CustomAwaitable):
939+
for action in ('complete', 'close', 'throw'):
940+
with self.subTest(factory=factory, action=action):
941+
calls = []
942+
943+
def get_awaitable():
944+
wrapped = factory()
945+
calls.append(wrapped)
946+
return wrapped
947+
948+
wrapper = anext(aiter(get_awaitable, object()))
949+
try:
950+
self.assertIsNone(wrapper.aw_wrapped)
951+
self.assertEqual(calls, [])
952+
with self.assertRaises(AttributeError):
953+
wrapper.aw_wrapped = None
954+
with self.assertRaises(AttributeError):
955+
del wrapper.aw_wrapped
956+
self.assertEqual(next(wrapper), ('result',))
957+
self.assertIs(wrapper.aw_wrapped, calls[0])
958+
if factory is produce:
959+
delegate = calls[0].cr_await
960+
self.assertEqual(wrapper.send(None), ('result',))
961+
self.assertIs(wrapper.aw_wrapped, calls[0])
962+
self.assertIsNot(calls[0].cr_await, delegate)
963+
if action == 'complete':
964+
with self.assertRaises(StopIteration):
965+
wrapper.send(None)
966+
elif action == 'throw':
967+
with self.assertRaises(AwaitException):
968+
wrapper.throw(AwaitException)
969+
else:
970+
wrapper.close()
971+
self.assertIs(wrapper.aw_wrapped, calls[0])
972+
finally:
973+
wrapper.close()
974+
975+
def test_aiter_callable_wrapped_sentinel(self):
976+
async def produce():
977+
return 1
978+
979+
iterator = aiter(produce, 1)
980+
wrapper = anext(iterator)
981+
with self.assertRaises(StopAsyncIteration):
982+
wrapper.send(None)
983+
wrapped = wrapper.aw_wrapped
984+
self.assertIsNotNone(wrapped)
985+
self.assertEqual(inspect.getcoroutinestate(wrapped),
986+
inspect.CORO_CLOSED)
987+
wrapper.close()
988+
self.assertIs(wrapper.aw_wrapped, wrapped)
989+
exhausted = anext(iterator)
990+
with self.assertRaises(StopAsyncIteration):
991+
exhausted.send(None)
992+
self.assertIsNone(exhausted.aw_wrapped)
918993

919994
def test_aiter_callable_cancel(self):
920995
# Cancellation is delivered to the awaited callable result
@@ -931,9 +1006,15 @@ async def consume():
9311006
async def main():
9321007
task = asyncio.ensure_future(consume())
9331008
await asyncio.sleep(0)
1009+
wrapper = task.get_coro().cr_await
1010+
wrapped = wrapper.aw_wrapped
1011+
self.assertIsNotNone(wrapped)
9341012
task.cancel()
9351013
with self.assertRaises(asyncio.CancelledError):
9361014
await task
1015+
self.assertIs(wrapper.aw_wrapped, wrapped)
1016+
self.assertEqual(inspect.getcoroutinestate(wrapped),
1017+
inspect.CORO_CLOSED)
9371018
self.loop.run_until_complete(main())
9381019
self.assertEqual(cancelled, [1])
9391020

Lib/test/test_asyncio/test_graph.py

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -173,6 +173,51 @@ class FakeCoro:
173173

174174
self.assertEqual(len(result.call_stack), 2)
175175

176+
async def test_stack_aiter_callable(self):
177+
for nested in (False, True):
178+
with self.subTest(nested=nested):
179+
entered = asyncio.Event()
180+
blocked = asyncio.Event()
181+
182+
async def deep():
183+
entered.set()
184+
await blocked.wait()
185+
186+
async def produce():
187+
await deep()
188+
return None
189+
190+
async def worker():
191+
iterator = aiter(produce, None)
192+
if nested:
193+
iterator = aiter(iterator.__anext__, None)
194+
async for _ in iterator:
195+
pass
196+
197+
task = asyncio.create_task(worker(), name='worker')
198+
try:
199+
await entered.wait()
200+
stack, printed = capture_test_stack(fut=task)
201+
self.assertEqual(stack[:2], [
202+
'T<worker>',
203+
['a wait', 'a deep', 'a produce', 'a worker'],
204+
])
205+
for name in ('deep', 'produce', 'worker'):
206+
self.assertIn(f'.<locals>.{name}()', printed)
207+
for limit, names in (
208+
(0, []),
209+
(2, ['produce', 'worker']),
210+
(-2, ['wait', 'deep']),
211+
):
212+
graph = asyncio.capture_call_graph(task, limit=limit)
213+
self.assertEqual(
214+
[entry.frame.f_code.co_name
215+
for entry in graph.call_stack], names)
216+
finally:
217+
task.cancel()
218+
with self.assertRaises(asyncio.CancelledError):
219+
await task
220+
176221
async def test_stack_gather(self):
177222

178223
stack_for_deep = None
Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,2 @@
1+
Fix :func:`asyncio.print_call_graph` truncating the call stack at the awaitable
2+
returned by the callable form of :func:`aiter`.

Objects/iterobject.c

Lines changed: 20 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -805,7 +805,7 @@ acallawaitable_start(acallawaitableobject *aw)
805805
}
806806
return -1;
807807
}
808-
aw->aw_wrapped = awaitable;
808+
FT_ATOMIC_STORE_PTR_RELEASE(aw->aw_wrapped, awaitable);
809809
return 0;
810810
}
811811

@@ -932,6 +932,24 @@ acallawaitable_close(PyObject *op, PyObject *Py_UNUSED(dummy))
932932
return result;
933933
}
934934

935+
static PyObject *
936+
acallawaitable_get_wrapped(PyObject *op, void *Py_UNUSED(closure))
937+
{
938+
acallawaitableobject *aw = acallawaitableobject_CAST(op);
939+
PyObject *wrapped = FT_ATOMIC_LOAD_PTR_ACQUIRE(aw->aw_wrapped);
940+
if (wrapped == NULL) {
941+
Py_RETURN_NONE;
942+
}
943+
return Py_NewRef(wrapped);
944+
}
945+
946+
static PyGetSetDef acallawaitable_getset[] = {
947+
{"aw_wrapped", acallawaitable_get_wrapped, NULL,
948+
PyDoc_STR("Awaitable returned by the callable, or None before it is called."),
949+
NULL},
950+
{NULL}
951+
};
952+
935953
static PyMethodDef acallawaitable_methods[] = {
936954
{"send", acallawaitable_send, METH_O, send_doc},
937955
{"throw", acallawaitable_throw, METH_VARARGS, throw_doc},
@@ -958,4 +976,5 @@ PyTypeObject _PyACallIterAwaitable_Type = {
958976
.tp_iter = PyObject_SelfIter,
959977
.tp_iternext = acallawaitable_iternext,
960978
.tp_methods = acallawaitable_methods,
979+
.tp_getset = acallawaitable_getset,
961980
};

0 commit comments

Comments
 (0)