Asynchronous Processing Challenge with BentoML: Need Advice on Handling Concurrent Requests

[euremi]

I've developed multiple APIs using BentoML, and the general structure of my script looks like the following:

There are two functions, main1 and main2, which receive requests almost simultaneously. I'm aiming to achieve an asynchronous processing flow where the success message is received before the sub functions are fully executed. However, I'm facing an issue where main1 immediately receives the success message, but it doesn't wait for the completion of the sub1 function before main2 receives its success message.

Current Situation:

1. Requests made to main1 and main2
2. main1 receives "success1" message
3. sub1 function of main1 begins execution
4. main2 receives "success2" message
5. sub1 function of main1 completes execution
6. sub2 function of main2 completes execution

Desired Situation:

1. Requests made to main1 and main2
2. main1 receives "success1" message
3. sub1 function of main1 is executing + main2 receives "success2" message
4. sub1 function of main1 completes execution
5. sub2 function of main2 completes execution

Here is a simplified version of my code:

@svc.api(route='/execute/main1',
         input=bentoml.io.JSON(),
         output=bentoml.io.JSON())
async def main1():
    asyncio.ensure_future(sub1())
    return {'message': 'success1'}

@svc.api(route='/execute/main2',
         input=bentoml.io.JSON(),
         output=bentoml.io.JSON())
async def main2():
    asyncio.ensure_future(sub2())
    return {'message': 'success2'}

async def sub1():
    ...
    request.post()

async def sub2():
    ...
    requests.post()

If you have any references or advice that could help me achieve the desired behavior, I would greatly appreciate it! Thank you!

[aarnphm]

shouldn't it just be

@svc.api(route='/execute/main1',
         input=bentoml.io.JSON(),
         output=bentoml.io.JSON())
async def main1():
    await sub1()
    return {'message': 'success1'}

?

[frostming]

You need some synchronization primitives to communicate between the two coroutine main1 and main2.

But it seems more like an improper architecture design. Because both main1 and main2 are async endpoints that can receive multiple requests simultaneously. What if there are multiple sub1 running at the same time, which one should main2 wait for?

[KimSoungRyoul]

starlette support BackgroundTask
use starlette's feature with fastapi

import asyncio

import bentoml
import numpy as np
import torch
from fastapi import FastAPI
from starlette.responses import JSONResponse
from torch import Tensor

from starlette.background import BackgroundTask

runner = bentoml.pytorch.get("iris_clf:latest").to_runner()

svc = bentoml.Service(name="sample-dummy-bento", runners=[runner])

fastapi_app = FastAPI()
# fastapi app with bentoml Service
svc.mount_asgi_app(fastapi_app)


async def sub1( input_array: list[list[float]]):
    print("task1 start~~~~") 
    res = request.post() # highly recommend to use httpx async client instead of request
    print(f"sub1: {res}")


async def sub2( input_array: list[list[float]]):
    print("task2 start~~~~")
    res = request.post() # highly recommend to use httpx async client instead of request
    print(f"sub2: {res}")


@fastapi_app.post("/predict-fastapi")
def predict_fastapi(input_array: list[list[float]]):
    task = BackgroundTask(sub1, input_array=input_array)
    return JSONResponse("message": "success", background=task)

@fastapi_app.post("/predict2-fastapi")
def predict2_fastapi(input_array: list[list[float]]):
    task = BackgroundTask(sub2, runner=runner, input_array=input_array)
    return JSONResponse({"message":"success"}, background=task)

this issue was resolved in bentoml slack KR channel