Usando vários trabalhadores em uma tarefa em segundo plano - Fast-API

Aug 17 2020

Estou tentando processar um arquivo carregado pelo usuário. No entanto, quero que o usuário obtenha uma resposta assim que o upload for concluído e encerre a conexão, mas continue processando o arquivo. Portanto, estou usando BackgroundTasks.add_tasks e meu código se parece com isto:

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)

Infelizmente, o código acima só executa um por um. Além disso, tenho que esperar até que todos os resultados sejam processados e, em seguida, imprimir o comando em foo funciona, ou seja, digamos que eu tenha n linhas no csv, depois que todos os n forem processados é quando vejo os comandos de impressão para todos. Meu programa é executado em 20 trabalhadores, mas enquanto esse processo está em execução, ele utiliza apenas cerca de 1% da CPU (foo não é uma tarefa de computação, é mais uma tarefa de IO / limite de rede). Isso me faz pensar que o processo em segundo plano está sendo executado em apenas um trabalhador. Eu tentei ProcessPoolExecutor da seguinte maneira:

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)

No entanto, recebo o seguinte erro:

processpoolexecutor não pode conservar objeto local

Eu consegui superar esse erro mudando minha abordagem de:

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

para:

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

No entanto, recebo este erro:

Arquivo "uvloop / loop.pyx", linha 2658, em uvloop.loop.Loop.run_in_executor AttributeError: O objeto 'Loop' não tem o atributo 'enviar'

Para resumir: para processar uma linha, posso atingir o ponto de extremidade "/ foo" . Agora, quero processar um csv de 200 linhas. Portanto, primeiro eu aceito o arquivo do usuário, retorno uma mensagem de sucesso e encerro essa conexão. O csv é então adicionado a uma tarefa em segundo plano que deve mapear cada linha para o ponto de extremidade "/ foo" e fornecer os resultados de cada linha. No entanto, todas as abordagens que tentei até agora parecem estar usando apenas um thread e estão processando cada linha uma por uma. Eu gostaria de uma abordagem em que eu possa processar várias linhas juntas, quase como se estivesse atingindo o ponto de extremidade "/ foo" várias vezes ao mesmo tempo, como podemos usar ferramentas como o Apache JMeter.

Respostas

1 alex_noname Aug 18 2020 at 10:29

Você poderia fazer o processamento em paralelo sem usar o terminal. Abaixo está um exemplo simplificado (sem usar fooendpoint) com base em seu código: