Использование нескольких воркеров в фоновой задаче - Fast-API

Aug 17 2020

Я пытаюсь обработать файл, загруженный пользователем. Однако я хочу, чтобы пользователь получил ответ после завершения загрузки и разорвал соединение, но продолжил обработку файла. Поэтому я использую BackgroundTasks.add_tasks, и мой код выглядит примерно так:

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)

К сожалению, приведенный выше код выполняется только по очереди. Более того, мне нужно подождать, пока все результаты будут обработаны, а затем сработает оператор печати в foo , т.е. скажем, у меня есть n строк в csv, после того, как все n обработаны, это когда я вижу операторы печати для всех. Моя программа работает с 20 рабочими, но пока этот процесс работает, он использует только около 1% ЦП (foo не является вычислительной задачей, это скорее задача, связанная с вводом-выводом / сетью). Это наводит меня на мысль, что фоновый процесс работает только на 1 работнике. Я пробовал ProcessPoolExecutor следующим образом:

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)

Однако я получаю следующую ошибку:

processpoolexecutor не может обработать локальный объект

Мне удалось преодолеть эту ошибку, изменив свой подход с:

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]

Однако затем я получаю эту ошибку:

Файл "uvloop / loop.pyx", строка 2658, в uvloop.loop.Loop.run_in_executor AttributeError: объект 'Loop' не имеет атрибута 'submit'

Подводя итог: чтобы обработать одну строку, я могу попасть в конечную точку «/ foo» . Теперь я хочу обработать CSV из 200 строк. Итак, сначала я принимаю файл от пользователя, возвращаю сообщение об успешном завершении и разрываю это соединение. Затем csv добавляется к фоновой задаче, которая должна сопоставлять каждую строку с конечной точкой «/ foo» и давать мне результаты для каждой строки. Однако все подходы, которые я пробовал до сих пор, похоже, используют только один поток и обрабатывают каждую строку одну за другой. Мне нужен подход, при котором я мог бы обрабатывать несколько строк вместе, как если бы я несколько раз одновременно нажимал на конечную точку «/ foo», как мы можем использовать такие инструменты, как Apache JMeter.

Ответы

1 alex_noname Aug 18 2020 at 10:29

Вы можете выполнять обработку параллельно без использования конечной точки. Ниже приведен упрощенный пример (без использования fooконечной точки) на основе вашего кода: