How to use async request in BentoML?

[nyongja]

Hello, I have a question about the asyncio inside the bentoml.

Currently, my service requires an async function call from the predict function(for redis request).
To this end, I used asyncio.get_event_loop() and the following error occurred.
RuntimeError: There is no current event loop in thread 'AnyIO worker thread'.

I solved this problem using asyncio.new_event_loop() instead of asyncio.get_event_loop().
However, I wonder why asyncio.get_event_loop() does not work inside the bentoml service.

I also tried to call the predict function from the runner’s init function for warmup, where asyncio.get_event_loop() works well, but the following error occurs in loop.run_until_complete.
RuntimeError: this event loop is already running.

When running the Bentoml service, what influences the async function call?
Is there another appropriate way to call async functions in the bentoml service?

Please note that below is a simplified example of our code.

bento_service.py

class Runner(bentoml.Runnable):
    SUPPORTED_RESOURCES = "cpu"
    SUPPORTS_CPU_MULTI_THREADING = True

    def __init__(self):
        self.predictor = Predictor()
        self.warmup()

    def warmup(self):
        res = self.predictor.predict(5)
        return res

    @bentoml.Runnable.method(batchable=False)
    def predict(self, req: dict):
        res = self.predictor.predict(req)
        return res

runner = bentoml.Runner(
    Runner, name="test_runner"
)

svc = bentoml.Service(
    name="test_service", runners=[runner]
)

@svc.api(input=JSON(), output=JSON(), route="predict")
def predict(req: dict()) -> dict:
    res = runner.predict.run(req["num"])
    return {"res": res}

predictor.py

class Predictor:
    def __init__(self):
        self.test_var = 3

    def predict(self, num):
        try:
            loop = asyncio.get_event_loop()
        except RuntimeError:
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
        num = loop.run_until_complete(async_test(num))

        return self.test_var * num

async_util.py

async def async_test(num):
    await asyncio.sleep(1)
    return num + 1

[sauyon]

Hiya, I think what you're looking for is simply

@svc.api(...)
async def ...

[nyongja]

Thank you for reply. But that method doesn't solve the problem in the warmup code.🥲

[parano]

Here's related documentation on defining async endpoint: https://docs.bentoml.org/en/latest/concepts/service.html#sync-vs-async-apis

There's no need for get_event_loop because BentoML will help manage that.

For example, here's what you could do: https://github.com/bentoml/BentoML/blob/main/examples/quickstart/service.py

import numpy as np

import bentoml
from bentoml.io import NumpyNdarray

iris_clf_runner = bentoml.sklearn.get("iris_clf:latest").to_runner()

svc = bentoml.Service("iris_classifier", runners=[iris_clf_runner])


@svc.api(
    input=NumpyNdarray.from_sample(
        np.array([[4.9, 3.0, 1.4, 0.2]], dtype=np.double), enforce_shape=False
    ),
    output=NumpyNdarray.from_sample(np.array([0.0], dtype=np.double)),
)
async def classify(input_series: np.ndarray) -> np.ndarray:
    return await iris_clf_runner.predict.async_run(input_series)

You can invoke other async functions in the async API as well, e.g.:

import redis.asyncio as redis

r =  redis.Redis(...)

svc = bentoml.Service(...)

@svc.api(input=Text(), output=NumpyNdarray())
async def classify(user_id: str) -> np.ndarray:
    features = await r.get(user_id)
    return await model_runner.predict.async_run(features)