Cuando los usuarios usan mi aplicación web, me gustaría ejecutar tareas que demoran mucho en segundo plano.
La librería threading no ejecuta código realmente en paralelo por culpa del GIL. Cómo puedo ejecutar tareas en otro hilo?
A pesar del comentario que he puesto antes, voy a intentar dar un ejemplo usando Futures:
Los "futures" son una abstracción para la ejecución concurrente de código que igual sirven para la ejecución de hilos (Threads) que para la ejecución de procesos (Process). Para ello utiliza un gestor de ejecución que es quien se encarga de la ejecución y de entregar los resultados (las "promesas") a quien las solicite.
Por ejemplo, queremos calcular el factorial de una secuencia de números. Con
with executor as e:
futures = {e.submit(fact, i):i for i in range(0,4000)}
lanzaríamos la ejecución en paralelo del cálculo del factorial de 4000 números. El executor
es quien gestiona hilos (ThreadPoolExecutor
) o procesos (ProcessPoolExecutor
), incluso podemos dejar que decida el número de "workers" que debe emplear según sea el número de cores de nuestra CPU.
El ejemplo completo:
import time
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, wait
def fact(n):
res = 1
for i in range(2, n+1):
res *= i
return res
if __name__ == "__main__":
t0 = time.time()
with ThreadPoolExecutor(max_workers=50) as e:
fs = {e.submit(fact, i):i for i in range(4000)}
wait(fs)
t1 = time.time()
with ProcessPoolExecutor() as e:
fs = {e.submit(fact, i):i for i in range(4000)}
wait(fs)
t2 = time.time()
print(f"Ejecución con hilos: {t1-t0:.2f}s")
print(f"Ejecución con procesos: {t2-t1:.2f}s")
Por lo general, los tiempos en windows con procesos son bastante malos, pudiendo llegar a ser peores que con hilos en algunos casos.
Un modo de trabajar "asíncronamente" sería ir metiendo al ejecutor más tareas sobre la marcha, algo así:
import time
from collections import deque
from itertools import islice
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, as_completed
def fact(n):
res = 1
for i in range(2, n+1):
res *= i
return res
if __name__ == "__main__":
with ProcessPoolExecutor() as e:
# metemos 1000 procesos
fs1 = { e.submit(fact, i):i for i in range(1000) }
print("Lanzados 1000 futuros")
# obtenemos los primeros 10 resultados
res10 = { fs1[f]:f.result() for f in islice(as_completed(fs1), 10)}
# metemos otros 100000 más
fs2 = { e.submit(fact, i):i for i in range(1000, 100000) }
print("Lanzados 100000 futuros más")
# obtenemos 10 resultados de la segunda tanda
res20 = { fs2[f]:f.result() for f in islice(as_completed(fs2), 1000, 1010)}
# obtenemos los últimos 10 resultados de la tanda anterior
res30 = { fs1[f]:f.result() for f in deque(as_completed(fs1), 10)}
# Espera para comprobar cómo sube el consumo de CPU
print("Espera de 3 segundos")
time.sleep(3)
# Cancelamos todos los procesos que no hayan acabado de la segunda tanda
for f in fs2:
f.cancel()
print("Cierre del ejecutor de procesos")
print()
for (n,res) in res10.items():
print(f"fact({n:4d}) = {str(res)[:80]}...")
print()
for (n,res) in res20.items():
print(f"fact({n:4d}) = {str(res)[:80]}...")
print()
for (n,res) in res30.items():
print(f"fact({n:4d}) = {str(res)[:80]}...")
Edición: he simplificado el código para que se entienda algo mejor.
Hay que destacar que res20
almacena resultados únicamente de la segunda tanda de futuros futures2
. En ese instante podríamos haber cancelado la ejecución de los futuros pendientes de la primera tanda si no se iban a necesitar para nada más.
PD: si no te funciona, asegúrate que usas python 3.6
Una forma eficiente y escalable de ejecutar tareas de forma asíncrona es usar una librería de colas como celery. Con esta librería, definís "workers" que son procesos (no threads) que son los que ejecutan las tareas pesadas. Un aspecto interesante de esta solución es que pueden haber muchos workers (incluso en diferentes servidores) ejecutando las tareas.
La arquitectura de la solución es la siguiente:
La siguiente es una aplicación sencilla que usa redis
como Broker. consumidor.py
le envía mensajes a productor.py
Asumo que redis
está andando y que virtualenv
está instalado. Con estos comandos se pueden instalar todas las dependencias:
virtualenv env
source env/bin/activate
pip install celery[redis]
Tanto el consumidor como el productor necesitan la configuración, así que las voy a guardar en config.py
(estoy usando la base de datos 1
de mi redis
local):
config = {
'broker': 'redis://localhost:6379/1'
}
Este es el contendo de productor.py
. Sólo expone una tarea (ejecutar_tarea
) que demora 10 segundos antes de imprimir el resultado. Esto es para ver cómo esta demora afecta al consumidor:
from config import config
from celery import Celery
from time import sleep
app = Celery('tasks', broker=config['broker'])
# pasar la configuracion como json
app.conf.CELERY_ACCEPT_CONTENT = ['json']
app.conf.CELERY_TASK_SERIALIZER = 'json'
@app.task
def ejecutar_tarea(mensaje):
# esto demoooora
sleep(10)
print 'Mensaje recibido: %s' % mensaje
Y éste es el código de consumidor.py
. Lo único que hace es recibir un mensaje de la consola y enviarlo al productor:
from productor import ejecutar_tarea
while True:
mensaje = raw_input('Mensaje: ')
ejecutar_tarea.delay(mensaje)
La mejor forma de testear esto es en la consola, abrir dos y activar el entorno virtual.
El productor se ejecuta de esta forma:
celery -A productor worker -l info
acá lo estoy arrancando con nivel de log: info
para tener detalles de qué es lo que está pasando. Las distintas opciones que se pueden usar al iniciar un worker están detalladas en la documentación de celery.
El consumidor se ejecuta corriendo:
python consumidor.py
Al escribir un mensaje en el consumidor, se puede ver que el productor lo recibe y luego de 10 segundos lo imprime. Lo interesante es que el consumidor no necesita esperar esos 10 segundos, sino que está disponible instantáneamente para procesar otro. Si el productor recibe muchos mensajes, entonces se encolan.
En fin, esta estrategia es sencilla de configurar y funciona muy bien. Cuando hace falta más fuerza, es relativamente sencillo agregar nuevos workers que ayuden con la carga.
futures
) o intentar entender el móduloasyncio
. Yo hace tiempo que me cambié a otros lenguajes (funcionales) por este motivo y no sabría darte ahora un ejemplo en python.