백그라운드 작업에서 여러 작업자 사용-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 에서 문을 인쇄해야 합니다. 즉, csv에 n 줄이 있다고 가정 해 보겠습니다. 모든 n이 처리 된 후 모두에 대한 인쇄 문을 볼 때입니다. 내 프로그램은 20 명의 작업자에서 실행되지만이 프로세스가 실행되는 동안 CPU의 약 1 % 만 사용합니다 (foo는 계산 작업이 아니라 IO / 네트워크 바인딩 작업에 가깝습니다). 이것은 백그라운드 프로세스가 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" 끝점을 누를 수 있습니다 . 이제 200 줄의 csv를 처리하고 싶습니다. 그래서 먼저 사용자로부터 파일을 수락하고 성공 메시지를 반환하고 해당 연결을 종료합니다. csv가 백그라운드 작업에 추가되어 각 줄을 "/ foo" 끝점에 매핑하고 각 줄에 대한 결과를 제공해야합니다. 그러나 지금까지 시도한 모든 접근 방식은 하나의 스레드 만 사용하는 것으로 보이며 각 줄을 하나씩 처리하고 있습니다. 마치 Apache JMeter와 같은 도구를 사용하는 것처럼 "/ foo" 끝점을 동시에 여러 번 누르는 것처럼 여러 줄을 함께 처리 할 수있는 접근 방식을 원합니다 .

답변

1 alex_noname Aug 18 2020 at 10:29

끝점을 사용하지 않고 병렬로 처리 할 수 있습니다. 다음은 foo코드를 기반으로 한 간단한 예 ( 엔드 포인트 를 사용하지 않음 )입니다.