2

I want to caculate python asyncio performance and select only cpu time cost.

I have read how-to-measure-pythons-asyncio-code-performance.

But asyncio.Task._step was changed to private which wrote into c modules

import asyncio
print(asyncio.Task)  # _asyncio.Task

print(dir(asyncio.Task))
['__await__', '__class__', '__del__', '__delattr__', '__dir__', '__doc__', 
'__eq__', '__format__', '__ge__', '__getattribute__', '__gt__', '__hash__', 
'__init__', '__init_subclass__', '__iter__', '__le__', '__lt__', '__ne__', 
'__new__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__', 
'__str__', '__subclasshook__', '_asyncio_future_blocking', '_callbacks', '_coro', 
'_exception', '_fut_waiter', '_log_destroy_pending', '_log_traceback', '_loop', 
'_must_cancel', '_repr_info', '_result', '_source_traceback', '_state', 'add_done_callback', 
'all_tasks', 'cancel', 'cancelled', 'current_task', 'done', 'exception', 'get_loop', 
'get_stack', 'print_stack', 'remove_done_callback', 'result', 'set_exception', 'set_result']

I can't hack to use only py-code because I don't want to lose c code performance.

5
  • Why not use standard tools like time? Also see my edit :) Commented Apr 1, 2019 at 12:00
  • If you aren't doing something very special, you don't need to measure CPU time cost. What you really want is to make sure some coroutine doesn't block event loop for too long. To achieve it use debug mode of event loop. No warnings will mean everything is ok and CPU takes much less time than network I/O. Commented Apr 1, 2019 at 12:11
  • @Vincent time.process_time is good but it is not great enough in coroutine. When I use asyncio.gather the process time is not correct which will be affect by the other jobs. While the old way in Python3.6 is good because it only calculate muiti step in own task. Commented Apr 3, 2019 at 8:33
  • @tianjunma That makes sense. It's pretty hackish but I guess you could patch asyncio handles. Commented Apr 3, 2019 at 12:08
  • @Vincent Wheather is it safe, inherit from asyncio.tasks._PyTask to create a TimedTask factory? Commented Nov 15, 2019 at 11:40

1 Answer 1

0

Something like this could be simple and useful,

import time

def time_this(func, *args, **kwargs):
    st = time.process_time()
    func(*args, **kwargs)
    return time.process_time() - st

Below's an example,

In [1]: import time, asyncio
   ...: 
   ...: def time_this(func, *args, **kwargs):
   ...:     st = time.process_time()
   ...:     func(*args, **kwargs)
   ...:     return time.process_time() - st
   ...: 
   ...: print(time_this(asyncio.run, asyncio.sleep(5)))
0.001157819000000007
Update #1:

Or you could follow this approach, writing your own event loop policy.

Sign up to request clarification or add additional context in comments.

Comments

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.