การใช้คนงานหลายคนในงานเบื้องหลัง - Fast-API

Question

ฉันกำลังพยายามประมวลผลไฟล์ที่ผู้ใช้อัพโหลด อย่างไรก็ตามฉันต้องการให้ผู้ใช้ได้รับคำตอบเมื่อการอัปโหลดเสร็จสิ้นและยุติการเชื่อมต่อ แต่ยังคงดำเนินการประมวลผลไฟล์ต่อไป ดังนั้นฉันจึงใช้ 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 คน แต่ในขณะที่กำลังทำงานอยู่จะใช้ CPU ประมาณ 1% เท่านั้น (foo ไม่ใช่งานคำนวณ แต่เป็นงานที่เชื่อมโยงกับ IO / Network) ทำให้ฉันคิดว่ากระบวนการเบื้องหลังกำลังทำงานกับผู้ปฏิบัติงาน 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

Answer 1

คุณสามารถทำการประมวลผลแบบขนานโดยไม่ต้องใช้จุดสิ้นสุด ด้านล่างนี้เป็นตัวอย่างที่เรียบง่าย (โดยไม่ต้องใช้fooปลายทาง) ตามรหัสของคุณ:

import asyncio
import sys
import uvicorn
from fastapi import FastAPI, BackgroundTasks, UploadFile, File
from loguru import logger


logger.remove()
logger.add(sys.stdout, colorize=True, format="<green>{time:HH:mm:ss}</green> | {level} | <level>{message}</level>")

app = FastAPI()


async def async_io_bound(line: str):
    await asyncio.sleep(3)  # Pretend this is IO operations
    return f"Line '{line}' processed"


async def process(csv):
    """ Processing CSV and generate data"""
    tasks = [async_io_bound(line) for line in csv]
    logger.info("start processing")
    result = await asyncio.gather(*tasks)
    for i in result:
        logger.info(i)


@app.post("/upload-to-process")
async def upload(background_tasks: BackgroundTasks, csv: UploadFile = File(...)):
    background_tasks.add_task(process, csv.file)
    return {"result": "CSV has been uploaded successfully"}

if __name__ == "__main__":
    uvicorn.run("app3:app", host="localhost", port=8001)

ตัวอย่างของเอาต์พุต (ทุกบรรทัดถูกประมวลผลแบบขนาน):

INFO:     ::1:52358 - "POST /upload-to-process HTTP/1.1" 200 OK
13:21:31 | INFO | start processing
13:21:34 | INFO | Line 'b'one, two\n'' processed
13:21:34 | INFO | Line 'b'0, 1\n'' processed
13:21:34 | INFO | Line 'b'1, 1\n'' processed
13:21:34 | INFO | Line 'b'2, 1\n'' processed
13:21:34 | INFO | Line 'b'3, 1\n'' processed
13:21:34 | INFO | Line 'b'4, 1\n'' processed
13:21:34 | INFO | Line 'b'5, 1\n'' processed
13:21:34 | INFO | Line 'b'6, 1\n'' processed
13:21:34 | INFO | Line 'b'7, 1\n'' processed
13:21:34 | INFO | Line 'b'8, 1\n'' processed
13:21:34 | INFO | Line 'b'9, 1\n'' processed