Jak mierzyć wydajność asynuatu w Pythonie?

Question

Nie mogę używać zwykłych narzędzi i technik do mierzenia wydajności koroutynu, ponieważ nie należy brać pod uwagę czasu, jaki zajmuje await (lub powinien on po prostu wziąć pod uwagę czas potrzebny na odczyt z wyczekiwanej, ale nie opóźnionej IO) .

Jak zmierzyć czas, jaki zajmuje coroutine? Jak porównać 2 wdrożenia i znaleźć bardziej wydajne? Z jakich narzędzi korzystam?

Answer 1

Jednym ze sposobów jest poprawka loop._selector.select w celu zaoszczędzenia czasu i zapisania wszystkich operacji we/wy. Można to zrobić za pomocą menedżera kontekstowego:

@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 

następnie użyć innego menedżera kontekstowe do czasu pełnego biegu, i obliczyć różnicę między całkowitym czasem a czasem 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)) 

Oto prosty Przykład:

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

drukuje następujące sprawozdanie:

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

---

EDIT

Innym podejściem jest podklasy Task i zastąpić metodę _step do czasu wykonanie kroku:

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 

Jest więc możliwe, aby zarejestrować podklasę jako domyślny fabryka zadań:

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

sam przykład:

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

z wyjściem:

Result: 3 
CPU time: 0.0002 

Answer 2

Jeśli chcesz tylko do pomiaru wydajności „Twój” kod, można stosować podobne podejście do testów jednostkowych - tylko małpa poprawkę (nawet patch + Mock) najbliższy IO coroutine z Future of expected result.

Główną wadą jest to, że np. Klient http jest dość prosty, ale powiedzmy, momoko (pg client) ... może być trudny do zrobienia bez znajomości jego elementów wewnętrznych, nie obejmie narzutu biblioteki.

pro są tak samo jak w zwykłym testów:

• jest to łatwe do wdrożenia,
• mierzy coś), głównie własnej implementacji bez obciążania bibliotek zewnętrznych,
• testy wydajności są izolowane , łatwe do re-run,
• to uruchomić z wielu ładunków