Utilizzo di più lavoratori in un'attività in background - Fast-API

Aug 17 2020

Sto cercando di elaborare un file caricato dall'utente. Tuttavia, desidero che l'utente riceva una risposta una volta completato il caricamento e termini la connessione, ma continui a elaborare il file. Pertanto sto usando BackgroundTasks.add_tasks e il mio codice ha un aspetto simile a questo:

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)

Sfortunatamente, il codice sopra viene eseguito solo uno alla volta. Inoltre, devo aspettare che tutti i risultati siano elaborati e quindi l'istruzione print in foo funziona, cioè diciamo che ho n righe nel csv, dopo che tutte le n sono state elaborate è quando vedo le istruzioni print per tutti. Il mio programma funziona su 20 worker ma, mentre questo processo è in esecuzione, utilizza solo l'1% circa della CPU (foo non è un'attività di calcolo, è più un'attività legata a IO / rete). Questo mi fa pensare che il processo in background sia in esecuzione su 1 solo worker. Ho provato ProcessPoolExecutor come segue:

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)

Tuttavia, ottengo il seguente errore:

processpoolexecutor non può decapare l'oggetto locale

Sono riuscito a superare quell'errore cambiando il mio approccio da:

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

per:

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

Tuttavia, allora ottengo questo errore:

File "uvloop / loop.pyx", riga 2658, in uvloop.loop.Loop.run_in_executor AttributeError: L'oggetto "Loop" non ha l'attributo "submit"

Per riassumere: per elaborare una riga, posso premere il punto finale "/ foo" . Ora, voglio elaborare un csv di 200 righe. Quindi prima accetto il file dall'utente e restituisco un messaggio di successo e interrompo la connessione. Il csv viene quindi aggiunto a un'attività in background che dovrebbe mappare ogni riga all'endpoint "/ foo" e fornirmi i risultati per ogni riga. Tuttavia, tutti gli approcci che ho provato finora sembrano utilizzare un solo thread e stanno elaborando ogni riga uno per uno. Vorrei un approccio in cui posso elaborare più righe insieme, quasi come se stessi colpendo l' endpoint "/ foo" più volte contemporaneamente, come possiamo usare strumenti come Apache JMeter.

Risposte

1 alex_noname Aug 18 2020 at 10:29

È possibile eseguire l'elaborazione in parallelo senza utilizzare l'endpoint. Di seguito è riportato un esempio semplificato (senza utilizzare l' fooendpoint) basato sul codice: