How to measure Python's asyncio code performance? -

i can't use normal tools , technics measure performance of coroutine because time takes @ await should not taken in consideration (or should consider overhead of reading awaitable not io latency).

so how measure time coroutine takes ? how compare 2 implementations , find more efficent ? tools use ?

one way patch loop._selector.select in order time , save io operations. can done using context manager:

@contextlib.contextmanager def patch_select(*, loop=none):     if loop none:         loop = asyncio.get_event_loop()     old_select = loop._selector.select     # define new select method, used context     def new_select(timeout):         if timeout == 0:             return old_select(timeout)         start = time.time()         result = old_select(timeout)         total = time.time() - start         new_select.iotime += total         return result     new_select.iotime = 0.0     # patch select method     try:         loop._selector.select = new_select         yield new_select     finally:         loop._selector.select = old_select 

then use context manager time full run, , compute difference between total time , io time:

@contextlib.contextmanager def timeit(*, loop=none):     start = time.time()     patch_select() context:         yield     total = time.time() - start     io_time = context.iotime         print("io time: {:.3f}".format(io_time))     print("cpu time: {:.3f}".format(total - io_time))     print("total time: {:.3f}".format(total)) 

here simple example:

loop = asyncio.get_event_loop() timeit(loop=loop):     coro = asyncio.sleep(1, result=3)     result = loop.run_until_complete(coro)     print("result: {}".format(result)) 

it prints following report:

result: 3 io time: 1.001 cpu time: 0.011 total time: 1.012 

---

edit

another approach subclass task , override _step method time execution of step:

class timedtask(asyncio.task):      self.cputime = 0.0      def _step(self, *args, **kwargs):         start = time.time()         result = super()._step(*args, **kwargs)         self.cputime += time.time() - start         return result 

it possible register subclass default task factory:

loop = asyncio.get_event_loop() task_factory = lambda loop, coro: timedtask(coro, loop=loop) loop.set_task_factory(task_factory) 

same example:

coro = asyncio.sleep(1, result=3, loop=loop) task = asyncio.ensure_future(coro, loop=loop) result = loop.run_until_complete(task) print("result: {}".format(result)) print("cpu time: {:.4f}".format(task.cputime)) 

with output:

result: 3 cpu time: 0.0002