Pythonのasyncioコードのパフォーマンスを測定する方法は？

Question

コルーチンのパフォーマンスを測定するのに、awaitにかかる時間を考慮してはならないので、コルーチンのパフォーマンスを測定することはできません（または、IO待ち時間ではなく待ち時間を読み込むオーバーヘッドを考慮する必要があります） 。

コルーチンがかかる時間をどのように測定するのですか？ 2つの実装をどのように比較し、より効率的なものを見つけるか？どのツールを使用しますか？

Answer 1

1つの方法は、すべてのIO操作を実行して保存するために、loop._selector.selectにパッチを適用することです。これは、コンテキストマネージャを使用して行うことができます。

@contextlib.contextmanager 
def patch_select(*, loop=None): 
    if loop is None: 
     loop = asyncio.get_event_loop() 
    old_select = loop._selector.select 
    # Define the new select method, used as a 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 the select method 
    try: 
     loop._selector.select = new_select 
     yield new_select 
    finally: 
     loop._selector.select = old_select 

は、その後の時間に完全な実行を別のコンテキストマネージャを使用して、合計時間とIOの時間との差を計算する：ここで

@contextlib.contextmanager 
def timeit(*, loop=None): 
    start = time.time() 
    with patch_select() as 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)) 

は簡単です。例：

Result: 3 
IO time: 1.001 
CPU time: 0.011 
Total time: 1.012 

：

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

それは次のレポートを印刷しますデフォルトとして、サブクラスを登録することが可能である

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 

：

---

EDIT

別のアプローチは、Taskをサブクラス化し、時間にステップの実行を_stepメソッドをオーバーライドすることですタスクファクトリ：

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

同じ例：

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)) 

出力付：

Result: 3 
CPU time: 0.0002 

Answer 2

あなただけの「あなた」のコードのパフォーマンスを測定したい場合は、ユニットテストと同様のアプローチを使用でき - ちょうど猿パッチを（たとえパッチ+モック）を使用して、最も近いIOコルーチンをFutureの期待結果に置き換えます。主な欠点は、例えば、 httpクライアントはかなりシンプルですが、momoko（pgクライアント）というと、内部を知らずに行うのは難しいかもしれません。ライブラリのオーバーヘッドは含まれません。

はプロだけで普通のテストでのようなものです：

• それが何かを測定
• 、実装するのは簡単です;）、サードパーティのライブラリのオーバーヘッドなしにほとんど1の実装、
• パフォーマンステストは単離されています、
• 、再実行に簡単にそれが動作しますので、私はupvoteよ、多くのペイロード