1

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?

3
  • Una cosa es la ejecución en paralelo, con tantos procesos como cores tenga tu CPU, y otra cosa distinta es lo que desearías de responder a tantos usuarios como se pueda de una manera asíncrona, repartiendo mejor el tiempo entre cada uno. Para multiproceso, emplea el módulo multiprocessing o el concurrent. Para asíncronismo, el módulo asyncio o cualquiera de los frameworks que existen para ello. Commented el 10 may. 2017 a las 0:51
  • @ChemaCortes si pusieras una respuesta explicando eso, con algún ejemplo sería genial :)
    – g3rv4
    Commented el 10 may. 2017 a las 13:00
  • La pregunta está formulada sobre "cómo ejecutar tareas asíncronas". El modelo de ejecución de python es básicamente monohilo y tiene más incovenientes quitarle el GIL que dejarlo como está. Como mucho puedes usar varios procesos (ejemplo con futures) o intentar entender el módulo asyncio. Yo hace tiempo que me cambié a otros lenguajes (funcionales) por este motivo y no sabría darte ahora un ejemplo en python. Commented el 10 may. 2017 a las 14:28

2 respuestas 2

2

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

2

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:

  • Consumidor: la aplicación que usan los usuarios, si es web puede ser una aplicación django o flask
  • Productor: el "worker" que es el que hace la tarea pesada
  • Broker: es el mecanismo que usa el consumidor para almacenar los "trabajos pendientes" y es lo que "despierta" al productor cuando hay trabajo por hacer
  • Backend: es el mecanismo que usa el productor para almacenar los resultados de la tarea (si el consumidor no necesita saber el resultado, puede no configurarse)

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 virtualenvestá 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.

Tu Respuesta

By clicking “Publica tu respuesta”, you agree to our terms of service and acknowledge you have read our privacy policy.

¿No es la respuesta que buscas? Examina otras preguntas con la etiqueta o formula tu propia pregunta.