Async support with custom runner

[chlee1016]

Hi team!
I am trying to implement preprocessing runner (I/O runner) to download some feature from url.
However, there doesn't seem to be enough explanation for implementing async and custom runners.
(https://docs.bentoml.org/en/latest/concepts/api_io_descriptors.html)

I tried the implementation as below, but I'm not sure what part I'm missing.
I would appreciate it if you could let me know the missing point.
(It would be helpful to others who want to implement async and custom runner.)

1. simple code implementation

# bentoml_service.py
class PredictorRequest(BaseModel):
    """This class represents Predictor's request data type."""

    url: str


class PreRunner(Runner):
    def _setup(self):
        pass

    async def _download_features(self, url):
        async with aiohttp.ClientSession() as session:
            async with session.post(url) as response:
                return await response.read()

    async def _run_batch(self, list_of_request):
        results = await asyncio.gather(*(self._download_features(url=request.url) for request in list_of_request))
        return results

pre_runner = PreRunner('pre_runner')

service = MyService('my_service', runners=[pre_runner])

@service.api(input=JSON(pydantic_model=PredictorRequest), output=JSON())
async def predict(req: PredictorRequest) -> t.Dict:
    feature = await pre_runner.async_run_batch(req)
    return {'feature': feature}

2. Error log

(Click here to expand)

3. related issue
The issue below describes well about a rough example of async runner with decorator, but I want to know how can I do the same implementation without the decorator.
#2248

[timliubentoml]

hi @chlee1016! What is the data that you're passing to the predict function? Are you passing the correct data "json" to the endpoint?

[chlee1016]

hi @timliubentoml! Sorry for the confusion. I have additionally described pydantic model, the pydantic model will be passed to the predict function.

[chlee1016]

hello @timliubentoml !
I'd like to make the question a bit simpler.
Now the Runner instances exposes run and run_batch method, which will eventually be
piped to the _run_batch implementation of the Runner.

BentoML/bentoml/_internal/runner/runner.py

Lines 151 to 152 in d37aa9e

	Runner instances exposes `run` and `run_batch` method, which will eventually be
	piped to the `_run_batch` implementation of the Runner. BentoML applies dynamic

Does the user only need to implement the Runner class's def _run_batch(not async def _run_batch) when using async_run or async_run_batch?