How can I measure the length of an asyncio event loop?

Question

I have an application which is calling loop.run_forever() after scheduling some tasks. Those tasks will connect to web services and then schedule new tasks on the loop based on input from those services.

I want to find a way of keeping track of the loop to check whether there are tasks being created which are never completing. Ideally I would measure the number of tasks in the loop periodically and write it to a file, or make it available via an http call.

Answer 1

You can collect all tasks, then count them, compute some other metric of "loop length" or perform inspection.

asyncio.Task.all_tasks(loop=loop)