Reputation: 3773
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.
Upvotes: 2
Views: 1593
Reputation: 7383
In python 3.10, you can get the number of tasks in the current thread like that :
len(asyncio.all_tasks(asyncio.get_running_loop()))
Upvotes: 3
Reputation: 425
You can collect all tasks, then count them, compute some other metric of "loop length" or perform inspection.
asyncio.Task.all_tasks(loop=loop)
Upvotes: 3