Usar varios trabajadores en una tarea en segundo plano - Fast-API

Aug 17 2020

Estoy intentando procesar un archivo subido por el usuario. Sin embargo, quiero que el usuario obtenga una respuesta una vez que se complete la carga y finalice la conexión, pero continúe procesando el archivo. Por lo tanto, estoy usando BackgroundTasks.add_tasks y mi código se parece a esto:

class Line(BaseModel):
    line: str

@app.post("/foo")
async def foo(line: Line):
""" Processing line generate results"""

    ...

    result = ... # processing line.line
    print(results)
    return results

@app.post("/upload")
async def upload(background_tasks: BackgroundTasks, csv: UploadFile = File(...)):

    background_tasks.add_task(process, csv)
    return response.text("CSV has been uploaded successfully")


async def process(csv):
    """ Processing CSV and generate data"""

    tasks = [foo(line) for line in csv]
    result = await asyncio.gather(*tasks)

Desafortunadamente, el código anterior solo se ejecuta uno por uno. Además, tengo que esperar hasta que se procesen todos los resultados y luego imprimir la declaración en foo funciona, es decir, digamos que tengo n líneas en el csv, después de que se procesan todas las n es cuando veo las declaraciones de impresión para todas. Mi programa se ejecuta en 20 trabajadores, pero mientras este proceso se está ejecutando, solo utiliza alrededor del 1% de la CPU (foo no es una tarea de cálculo, es más una tarea vinculada a IO / Red). Esto me hace pensar que el proceso en segundo plano se está ejecutando solo en 1 trabajador. Probé ProcessPoolExecutor de la siguiente manera:

loop = asyncio.get_event_loop()
lines = [line_0, line_1, ..., line_n] # Extracted all lines from CSV
with ProcessPoolExecutor() as executor:
    results = [loop.run_in_executor(executor, lambda: foo(line)) for line in lines]
    results = loop.run_until_complete(*results)

Sin embargo, aparece el siguiente error:

processpoolexecutor no puede encurtir objetos locales

Me las arreglé para superar ese error cambiando mi enfoque de:

results = [loop.run_in_executor(executor, lambda: foo(line)) for line in lines]

results = [asyncio.ensure_future(foo(line=Line(line)) for line in lines]

Sin embargo, aparece este error:

Archivo "uvloop / loop.pyx", línea 2658, en uvloop.loop.Loop.run_in_executor AttributeError: El objeto 'Loop' no tiene atributo 'submit'

Para resumir: para procesar una línea, puedo presionar el punto final "/ foo" . Ahora, quiero procesar un csv de 200 líneas. Entonces, primero acepto el archivo del usuario y devuelvo un mensaje de éxito y finalizo esa conexión. Luego, el csv se agrega a una tarea en segundo plano que debe asignar cada línea al punto final "/ foo" y darme los resultados de cada línea. Sin embargo, todos los enfoques que he probado hasta ahora parecen estar usando solo un hilo y están procesando cada línea una por una. Me gustaría un enfoque en el que pueda procesar varias líneas juntas, casi como si estuviera presionando el punto final "/ foo" varias veces simultáneamente, como podemos usar herramientas como Apache JMeter.

Respuestas

1 alex_noname Aug 18 2020 at 10:29

Puede realizar el procesamiento en paralelo sin utilizar el punto final. A continuación, se muestra un ejemplo simplificado (sin usar el foopunto final) basado en su código: