Comment mesurer la performance du code asyncio de Python?

Question

Je ne peux pas utiliser les outils et les techniques normales pour mesurer la performance d'une coroutine, car le temps pris à await ne doit pas être pris en compte (ou seulement la latence de lecture de la latence attendue mais pas de l'IO) .

Alors, comment mesurer le temps que prend un coroutine? Comment puis-je comparer 2 implémentations et trouver le plus efficace? Quels outils est-ce que j'utilise?

Answer 1

L'une des façons est de corriger le loop._selector.select afin de chronométrer et de sauvegarder toutes les opérations d'E/S. Cela peut être fait en utilisant un gestionnaire de contexte:

@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 

Ensuite, utilisez un autre gestionnaire de contexte à temps l'exécution complète, et on calcule la différence entre le temps total et le temps 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)) 

Voici un simple exemple:

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

Il imprime le rapport suivant:

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

---

EDIT

Une autre approche consiste à sous-classe Task et remplacer la méthode _step à l'heure de l'exécution de l'étape consistant à:

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 

Il est alors possible d'enregistrer la sous-classe par défaut usine de tâche:

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

Même exemple:

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

Avec la sortie:

Result: 3 
CPU time: 0.0002 

Answer 2

Si vous ne souhaitez que pour mesurer la performance de « votre » code, vous pouvez utiliser approche similaire pour les tests unitaires - juste singe-patch (même patch + Mock) la plus proche IO coroutine avec Future du résultat attendu.

L'inconvénient principal est que par ex. Le client http est assez simple, mais disons momoko (client pg) ... il pourrait être difficile de faire sans connaître ses internes, cela n'inclura pas les frais généraux de la bibliothèque.

Le pro sont comme dans les tests ordinaires:

• il est facile à mettre en œuvre,
• il mesure quelque chose;), la plupart du temps sa mise en œuvre sans frais généraux des bibliothèques de tiers,
• tests de performance sont isolés , facile à Réexécutez,
• il est de fonctionner avec de nombreuses charges utiles