Patch loop runner _run_once

import asyncio
def _run_once(self):
	num_tasks = len(self._scheduled)
    print("num tasks in queue: {}".format(num_tasks))
    super(asyncio.SelectorEventLoop, self)._run_once()

EventLoop = asyncio.SelectorEventLoop
EventLoop._run_once = _run_once
loop = EventLoop()
asyncio.set_event_loop(loop)
async def task(n):
    await asyncio.sleep(n)
    print("sleep: {} sec".format(n))

coro = loop.create_task(task(3))
loop.run_until_complete(coro)
# num tasks in queue: 0
# num tasks in queue: 1
# num tasks in queue: 0
# sleep: 3 sec
# num tasks in queue: 0
loop.close()

Source: pythonsheets.com

Similar codes

0 vote136 views1 answer
What will be the output of the following program?
Python
46 votes147 views6 answers
Round to decimal places python
Python
1 vote147 views1 answer
Python show only 1st element of nested lists
Python
0 vote152 views1 answer
Gunicorn django static files
Python
0 vote141 views1 answer
Python loc id in list
Python
0 vote149 views1 answer
‘A’, ‘Q’, ‘BM’, ‘BA’, ‘BQ’ meaning in resample
Python
1 vote224 views1 answer
Printing with colors
Python
44 votes178 views2 answers
Repl.it python
Python
0 vote151 views1 answer
Randomly splits this DataFrame with the provided weights
Python
4 votes161 views2 answers
Levenshtein distance
Python