バックグラウンドタスクで複数のワーカーを使用する-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)

残念ながら、上記のコードは1つずつしか実行されません。さらに、すべての結果が処理されるまで待ってから、fooのprintステートメントが機能する必要があります。つまり、すべてのnが処理された後、すべてのprintステートメントが表示されたときに、csvに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はローカルオブジェクトをpickle化できません

私は自分のアプローチを次のように変更することで、なんとかそのエラーを克服することができました。

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.Loop.run_in_executorのファイル "uvloop / loop.pyx"、行2658 AttributeError： 'Loop'オブジェクトに属性 'submit'がありません

要約すると、1行を処理するには、「/ foo」エンドポイントに到達します。ここで、200行のcsvを処理したいと思います。したがって、最初にユーザーからファイルを受け取り、成功メッセージを返し、その接続を終了します。次に、csvがバックグラウンドタスクに追加され、各行が「/ foo」エンドポイントにマップされ、各行の結果が表示されます。ただし、これまでに試したすべてのアプローチは、1つのスレッドのみを使用しているようで、各行を1つずつ処理しています。Apache JMeterのようなツールを使用できるように、「/ foo」エンドポイントを同時に複数回ヒットするかのように、複数の行を一緒に処理できるアプローチが必要です。

回答

1 alex_noname Aug 18 2020 at 10:29

エンドポイントを使用せずに並行して処理を行うことができます。以下は、fooコードに基づいた（エンドポイントを使用しない）簡略化された例です。