Skip to content

Instantly share code, notes, and snippets.

@gvanrossum
Created September 24, 2022 04:51
Show Gist options
  • Save gvanrossum/08834f4212210ecdffe4d2079ae2f27a to your computer and use it in GitHub Desktop.
Save gvanrossum/08834f4212210ecdffe4d2079ae2f27a to your computer and use it in GitHub Desktop.
get awaiter stack in pure Python
import asyncio
from asyncio.tasks import Task
def _get_coro_awaiter(coro):
if hasattr(coro, "cr_await"):
awaiter = coro.cr_await
if hasattr(awaiter, "cr_await"):
return awaiter
return None
def get_await_stack(coro):
"""Return the chain of coroutines reachable from coro via its awaiter"""
stack = []
awaiter = _get_coro_awaiter(coro)
while awaiter is not None:
stack.append(awaiter)
awaiter = _get_coro_awaiter(awaiter)
return stack
async def f1():
return await f2()
async def f2():
# Create a new task for f3()
return await asyncio.ensure_future(f3())
# return await f3()
async def f3():
return await f3a()
async def f3a():
return await f4()
def dump_task_stack():
print("Stack retrieved from Task.get_stack():")
stack = asyncio.current_task().get_stack()
for frame in stack:
print(f" {frame.f_code.co_name}")
async def dump_await_stack():
print("Stack retrieved using awaiters:")
coro = asyncio.current_task()._coro
async def dummy_coro():
stack = [coro] + get_await_stack(coro)[:-1]
return stack
dummy_task = asyncio.create_task(dummy_coro())
stack = await dummy_task
for coro in reversed(stack):
print(f" {coro.cr_code.co_name}")
async def f4():
await asyncio.sleep(1)
dump_task_stack()
await dump_await_stack()
return 42
if __name__ == "__main__":
asyncio.run(f1())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment