¿Cómo medir el rendimiento del código asyncio de Python?

No puedo usar las herramientas y técnicas normales para medir el rendimiento de una coroutine porque el tiempo que se tarda en await no debe tomarse en consideración (o solo debe considerar la sobrecarga de la lectura de la latencia de IO esperada pero no la latencia).

Entonces, ¿cómo medir el tiempo que tarda una coroutine? ¿Cómo comparo 2 implementaciones y encuentro las más eficientes? ¿Qué herramientas utilizo?

Esta respuesta originalmente contenía dos soluciones diferentes: la primera se basaba en parches de mono y la segunda no funciona para Python 3.7 y posteriores. Esperamos que esta nueva versión presente un enfoque mejor y más robusto.

En primer lugar, las herramientas de tiempo estándar, como el tiempo, se pueden usar para determinar el tiempo de CPU de un progtwig, que suele ser lo que nos interesa al probar el rendimiento de una aplicación asíncrona. Esas mediciones también se pueden realizar en python usando la función time.process_time () :

 import time real_time = time.time() cpu_time = time.process_time() time.sleep(1.) sum(range(10**6)) real_time = time.time() - real_time cpu_time = time.process_time() - cpu_time print(f"CPU time: {cpu_time:.2f} s, Real time: {real_time:.2f} s") 

Vea a continuación la salida similar producida por ambos métodos:

 $ /usr/bin/time -f "CPU time: %U s, Real time: %es" python demo.py CPU time: 0.02 s, Real time: 1.02 s # python output CPU time: 0.03 s, Real time: 1.04 s # `time` output 

En una aplicación asyncio, puede suceder que alguna parte síncrona del progtwig termine realizando una llamada de locking, impidiendo efectivamente que el bucle de eventos ejecute otras tareas. Por lo tanto, es posible que queramos registrar por separado el tiempo que el bucle de eventos pasa esperando a partir del tiempo que toman otras tareas de IO.

Esto se puede lograr subclasificando el selector predeterminado para realizar alguna operación de sincronización y utilizando una política de bucle de eventos personalizada para configurar todo. Este fragmento de código proporciona dicha política junto con un administrador de contexto para imprimir diferentes métricas de tiempo.

 async def main(): print("~ Correct IO management ~") with print_timing(): await asyncio.sleep(1) sum(range(10**6)) print() print("~ Incorrect IO management ~") with print_timing(): time.sleep(0.2) await asyncio.sleep(0.8) sum(range(10**6)) print() asyncio.set_event_loop_policy(TimedEventLoopPolicy()) asyncio.run(main(), debug=True) 

Note la diferencia entre esas dos carreras:

 ~ Correct IO management ~ CPU time: 0.016 s Select time: 1.001 s Other IO time: 0.000 s Real time: 1.017 s ~ Incorrect IO management ~ CPU time: 0.016 s Select time: 0.800 s Other IO time: 0.200 s Real time: 1.017 s 

También tenga en cuenta que el modo de depuración de asyncio puede detectar esas operaciones de locking:

 Executing () created at ~/miniconda/lib/python3.7/asyncio/futures.py:288> took 0.243 seconds 

Si solo desea medir el rendimiento de “su” código, podría utilizar un método similar al de las pruebas unitarias: solo mono-parche (incluso parche + simulacro) la solución de IO más cercana con el futuro del resultado esperado.

El principal inconveniente es que, por ejemplo, el cliente http es bastante simple, pero digamos que momoko (pg client) … podría ser difícil de hacer sin conocer sus aspectos internos, no incluirá la sobrecarga de la biblioteca.

Los profesionales son como en las pruebas ordinarias:

  • es fácil de implementar,
  • mide algo;), en su mayoría implementación sin gastos generales de bibliotecas de terceros,
  • Las pruebas de rendimiento son aisladas, fáciles de volver a ejecutar,
  • es para correr con muchas cargas útiles