We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
fastapi 0.70 failing tests
When using `fastapi` latest version, some tests fail e.g. `test_infer`, if we use an older version of it (e.g. `0.68.2`) tests pass.

We need to take a deeper look just in case it uncovers a more subtle issue with asyncio (judging by the error we get).

here is the error for reference

```
est_infer[sum-model-None] _____________________________________

rest_client = <starlette.testclient.TestClient object at 0x7fa456254bd0>
inference_request = InferenceRequest(id=None, parameters=None, inputs=[RequestInput(name='input-0', shape=[1, 3], datatype='INT32', parameters=None, data=TensorData(__root__=[1, 2, 3]))], outputs=None)
model_name = 'sum-model', model_version = None

    @pytest.mark.parametrize(
        "model_name,model_version", [("sum-model", "v1.2.3"), ("sum-model", None)]
    )
    def test_infer(rest_client, inference_request, model_name, model_version):
        endpoint = f"/v2/models/{model_name}/infer"
        if model_version is not None:
            endpoint = f"/v2/models/{model_name}/versions/{model_version}/infer"
>       response = rest_client.post(endpoint, json=inference_request.dict())

tests/rest/test_endpoints.py:57: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/requests/sessions.py:590: in post
    return self.request('POST', url, data=data, json=json, **kwargs)
../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/starlette/testclient.py:484: in request
    json=json,
../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/requests/sessions.py:542: in request
    resp = self.send(prep, **send_kwargs)
../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/requests/sessions.py:655: in send
    r = adapter.send(request, **kwargs)
../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/starlette/testclient.py:266: in send
    raise exc
../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/starlette/testclient.py:263: in send
    portal.call(self.app, scope, receive, send)
../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/anyio/from_thread.py:230: in call
    return cast(T_Retval, self.start_task_soon(func, *args).result())
../../miniconda3/envs/test_nest_io/lib/python3.7/concurrent/futures/_base.py:435: in result
    return self.__get_result()
../../miniconda3/envs/test_nest_io/lib/python3.7/concurrent/futures/_base.py:384: in __get_result
    raise self._exception
../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/anyio/from_thread.py:177: in _call_func
    retval = await retval
../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/fastapi/applications.py:208: in __call__
    await super().__call__(scope, receive, send)
../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/starlette/applications.py:112: in __call__
    await self.middleware_stack(scope, receive, send)
../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/starlette/middleware/errors.py:181: in __call__
    raise exc
../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/starlette/middleware/errors.py:159: in __call__
    await self.app(scope, receive, _send)
../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/starlette/exceptions.py:82: in __call__
    raise exc
../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/starlette/exceptions.py:71: in __call__
    await self.app(scope, receive, sender)
../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/starlette/routing.py:656: in __call__
    await route.handle(scope, receive, send)
../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/starlette/routing.py:259: in handle
    await self.app(scope, receive, send)
../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/starlette/routing.py:61: in app
    response = await func(request)
mlserver/rest/app.py:24: in custom_route_handler
    return await original_route_handler(request)
../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/fastapi/routing.py:227: in app
    dependant=dependant, values=values, is_coroutine=is_coroutine
../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/fastapi/routing.py:159: in run_endpoint_function
    return await dependant.call(**values)
mlserver/rest/endpoints.py:52: in infer
    return await self._data_plane.infer(payload, model_name, model_version)
mlserver/handlers/dataplane.py:60: in infer
    prediction = await model.predict(payload)
mlserver/batching/hooks.py:43: in _inner
    return await batcher.predict(payload)
mlserver/batching/adaptive.py:36: in predict
    return await self._wait_response(internal_id)
mlserver/batching/adaptive.py:55: in _wait_response
    response = await async_response
mlserver/batching/adaptive.py:85: in _batcher
    async for batched in self._batch_requests():
mlserver/batching/adaptive.py:112: in _batch_requests
    read_op, timeout=timeout
../../miniconda3/envs/test_nest_io/lib/python3.7/asyncio/tasks.py:442: in wait_for
    return fut.result()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <Queue at 0x7fa456575490 maxsize=10 tasks=1>

    async def get(self):
        """Remove and return an item from the queue.
    
        If queue is empty, wait until an item is available.
        """
        while self.empty():
            getter = self._loop.create_future()
            self._getters.append(getter)
            try:
>               await getter
E               RuntimeError: Task <Task pending coro=<Queue.get() running at /home/sa/miniconda3/envs/test_nest_io/lib/python3.7/asyncio/queues.py:159> cb=[_release_waiter(<Future pendi...4566d8550>()]>)() at /home/sa/miniconda3/envs/test_nest_io/lib/python3.7/asyncio/tasks.py:392]> got Future <Future pending> attached to a different loop

../../miniconda3/envs/test_nest_io/lib/python3.7/asyncio/queues.py:159: RuntimeError
```


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `mlserver/batching/adaptive.py`
Content:
```
1 import time
2 import asyncio
3 
4 from asyncio import Future, Queue, wait_for, Task
5 from functools import partial
6 from typing import AsyncIterator, Awaitable, Dict, Tuple
7 
8 from ..model import MLModel
9 from ..types import (
10     InferenceRequest,
11     InferenceResponse,
12 )
13 from ..utils import generate_uuid
14 
15 from .requests import BatchedRequests
16 
17 
18 class AdaptiveBatcher:
19     def __init__(self, model: MLModel):
20         self._model = model
21 
22         self._max_batch_size = model.settings.max_batch_size
23         self._max_batch_time = model.settings.max_batch_time
24 
25         # Save predict function before it gets decorated
26         self._predict_fn = model.predict
27         self._requests: Queue[Tuple[str, InferenceRequest]] = Queue(
28             maxsize=self._max_batch_size
29         )
30         self._async_responses: Dict[str, Future[InferenceResponse]] = {}
31         self._batching_task = None
32 
33     async def predict(self, req: InferenceRequest) -> InferenceResponse:
34         internal_id, _ = await self._queue_request(req)
35         self._start_batcher_if_needed()
36         return await self._wait_response(internal_id)
37 
38     async def _queue_request(
39         self,
40         req: InferenceRequest,
41     ) -> Tuple[str, Awaitable[InferenceResponse]]:
42         internal_id = generate_uuid()
43         await self._requests.put((internal_id, req))
44 
45         loop = asyncio.get_running_loop()
46         async_response = loop.create_future()
47         self._async_responses[internal_id] = async_response
48 
49         return internal_id, async_response
50 
51     async def _wait_response(self, internal_id: str) -> InferenceResponse:
52         async_response = self._async_responses[internal_id]
53 
54         try:
55             response = await async_response
56             return response
57         finally:
58             del self._async_responses[internal_id]
59 
60     def _start_batcher_if_needed(self):
61         if self._batching_task is not None:
62             if not self._batching_task.done():
63                 # If task hasn't finished yet, let it keep running
64                 return
65 
66         self._batching_task = asyncio.create_task(self._batcher())
67         self._batching_task.add_done_callback(self._batching_task_callback)
68 
69     def _batching_task_callback(self, batching_task: Task):
70         err = batching_task.exception()
71         if err:
72             # Clear queue
73             self._clear_queue(err)
74 
75     def _clear_queue(self, err: BaseException):
76         # Cancel all pending async responses
77         for async_response in self._async_responses.values():
78             async_response.set_exception(err)
79 
80         # Empty queue
81         for _ in range(self._requests.qsize()):
82             self._requests.get_nowait()
83 
84     async def _batcher(self):
85         async for batched in self._batch_requests():
86             # We run prediction as a Task to ensure it gets scheduled
87             # immediately.
88             # That way, we can process multiple batches concurrently.
89             predict_task = asyncio.create_task(self._predict_fn(batched.merged_request))
90             predict_task.add_done_callback(partial(self._predict_callback, batched))
91 
92     def _predict_callback(self, batched: BatchedRequests, predict_task: Task):
93         try:
94             batched_response = predict_task.result()
95             responses = batched.split_response(batched_response)
96             for internal_id, response in responses.items():
97                 self._async_responses[internal_id].set_result(response)
98         except Exception as err:
99             for internal_id in batched.inference_requests.keys():
100                 self._async_responses[internal_id].set_exception(err)
101 
102     async def _batch_requests(self) -> AsyncIterator[BatchedRequests]:
103         while not self._requests.empty():
104             to_batch: Dict[str, InferenceRequest] = {}
105             start = time.time()
106             timeout = self._max_batch_time
107 
108             try:
109                 while len(to_batch) < self._max_batch_size:
110                     read_op = self._requests.get()
111                     internal_id, inference_request = await wait_for(
112                         read_op, timeout=timeout
113                     )
114                     to_batch[internal_id] = inference_request
115 
116                     # Update remaining timeout
117                     current = time.time()
118                     timeout = timeout - (current - start)
119             except asyncio.TimeoutError:
120                 # NOTE: Hit timeout, continue
121                 pass
122 
123             yield BatchedRequests(to_batch)
124
```

Path: `setup.py`
Content:
```
1 import os
2 
3 from typing import Dict
4 from setuptools import setup, find_packages
5 
6 ROOT_PATH = os.path.dirname(__file__)
7 PKG_NAME = "mlserver"
8 PKG_PATH = os.path.join(ROOT_PATH, PKG_NAME)
9 
10 
11 def _load_version() -> str:
12     version = ""
13     version_path = os.path.join(PKG_PATH, "version.py")
14     with open(version_path) as fp:
15         version_module: Dict[str, str] = {}
16         exec(fp.read(), version_module)
17         version = version_module["__version__"]
18 
19     return version
20 
21 
22 def _load_description() -> str:
23     readme_path = os.path.join(ROOT_PATH, "README.md")
24     with open(readme_path) as fp:
25         return fp.read()
26 
27 
28 setup(
29     name=PKG_NAME,
30     version=_load_version(),
31     url="https://github.com/SeldonIO/MLServer.git",
32     author="Seldon Technologies Ltd.",
33     author_email="hello@seldon.io",
34     description="ML server",
35     packages=find_packages(exclude=["tests", "tests.*"]),
36     install_requires=[
37         "click",
38         # We pin version of fastapi
39         # check https://github.com/SeldonIO/MLServer/issues/340
40         "fastapi==0.68.2",
41         "grpcio",
42         "importlib-metadata;python_version<'3.8'",
43         "numpy",
44         "pandas",
45         "protobuf",
46         "uvicorn",
47         "starlette_exporter",
48         "py-grpc-prometheus",
49     ],
50     extras_require={"all": ["orjson"]},
51     entry_points={"console_scripts": ["mlserver=mlserver.cli:main"]},
52     long_description=_load_description(),
53     long_description_content_type="text/markdown",
54     license="Apache 2.0",
55 )
56
```

Path: `mlserver/rest/errors.py`
Content:
```
1 from typing import Optional
2 from fastapi import Request, status
3 from pydantic import BaseModel
4 
5 from .responses import Response
6 from ..errors import MLServerError
7 
8 
9 class APIErrorResponse(BaseModel):
10     error: Optional[str] = None
11 
12 
13 def handle_mlserver_error(request: Request, exc: MLServerError) -> Response:
14     err_res = APIErrorResponse(error=str(exc))
15     return Response(status_code=status.HTTP_400_BAD_REQUEST, content=err_res.dict())
16 
17 
18 _EXCEPTION_HANDLERS = {MLServerError: handle_mlserver_error}
19
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/mlserver/batching/adaptive.py b/mlserver/batching/adaptive.py
--- a/mlserver/batching/adaptive.py
+++ b/mlserver/batching/adaptive.py
@@ -1,9 +1,13 @@
+# Required to deal with annotations including `Queue`
+# https://mypy.readthedocs.io/en/latest/common_issues.html#issues-with-code-at-runtime
+from __future__ import annotations
+
 import time
 import asyncio
 
 from asyncio import Future, Queue, wait_for, Task
 from functools import partial
-from typing import AsyncIterator, Awaitable, Dict, Tuple
+from typing import AsyncIterator, Awaitable, Dict, Optional, Tuple
 
 from ..model import MLModel
 from ..types import (
@@ -24,9 +28,7 @@
 
         # Save predict function before it gets decorated
         self._predict_fn = model.predict
-        self._requests: Queue[Tuple[str, InferenceRequest]] = Queue(
-            maxsize=self._max_batch_size
-        )
+        self.__requests: Optional[Queue[Tuple[str, InferenceRequest]]] = None
         self._async_responses: Dict[str, Future[InferenceResponse]] = {}
         self._batching_task = None
 
@@ -35,11 +37,21 @@
         self._start_batcher_if_needed()
         return await self._wait_response(internal_id)
 
+    @property
+    def _requests(self) -> Queue[Tuple[str, InferenceRequest]]:
+        # NOTE: We need to create Queue within the async request path (and not
+        # during __init__!!) to ensure that it shares the same AsyncIO loop.
+        if self.__requests is None:
+            self.__requests = Queue(maxsize=self._max_batch_size)
+
+        return self.__requests
+
     async def _queue_request(
         self,
         req: InferenceRequest,
     ) -> Tuple[str, Awaitable[InferenceResponse]]:
         internal_id = generate_uuid()
+
         await self._requests.put((internal_id, req))
 
         loop = asyncio.get_running_loop()
diff --git a/mlserver/rest/errors.py b/mlserver/rest/errors.py
--- a/mlserver/rest/errors.py
+++ b/mlserver/rest/errors.py
@@ -10,7 +10,7 @@
     error: Optional[str] = None
 
 
-def handle_mlserver_error(request: Request, exc: MLServerError) -> Response:
+async def handle_mlserver_error(request: Request, exc: MLServerError) -> Response:
     err_res = APIErrorResponse(error=str(exc))
     return Response(status_code=status.HTTP_400_BAD_REQUEST, content=err_res.dict())
 
diff --git a/setup.py b/setup.py
--- a/setup.py
+++ b/setup.py
@@ -35,9 +35,7 @@
     packages=find_packages(exclude=["tests", "tests.*"]),
     install_requires=[
         "click",
-        # We pin version of fastapi
-        # check https://github.com/SeldonIO/MLServer/issues/340
-        "fastapi==0.68.2",
+        "fastapi",
         "grpcio",
         "importlib-metadata;python_version<'3.8'",
         "numpy",

{"golden_diff": "diff --git a/mlserver/batching/adaptive.py b/mlserver/batching/adaptive.py\n--- a/mlserver/batching/adaptive.py\n+++ b/mlserver/batching/adaptive.py\n@@ -1,9 +1,13 @@\n+# Required to deal with annotations including `Queue`\n+# https://mypy.readthedocs.io/en/latest/common_issues.html#issues-with-code-at-runtime\n+from __future__ import annotations\n+\n import time\n import asyncio\n \n from asyncio import Future, Queue, wait_for, Task\n from functools import partial\n-from typing import AsyncIterator, Awaitable, Dict, Tuple\n+from typing import AsyncIterator, Awaitable, Dict, Optional, Tuple\n \n from ..model import MLModel\n from ..types import (\n@@ -24,9 +28,7 @@\n \n         # Save predict function before it gets decorated\n         self._predict_fn = model.predict\n-        self._requests: Queue[Tuple[str, InferenceRequest]] = Queue(\n-            maxsize=self._max_batch_size\n-        )\n+        self.__requests: Optional[Queue[Tuple[str, InferenceRequest]]] = None\n         self._async_responses: Dict[str, Future[InferenceResponse]] = {}\n         self._batching_task = None\n \n@@ -35,11 +37,21 @@\n         self._start_batcher_if_needed()\n         return await self._wait_response(internal_id)\n \n+    @property\n+    def _requests(self) -> Queue[Tuple[str, InferenceRequest]]:\n+        # NOTE: We need to create Queue within the async request path (and not\n+        # during __init__!!) to ensure that it shares the same AsyncIO loop.\n+        if self.__requests is None:\n+            self.__requests = Queue(maxsize=self._max_batch_size)\n+\n+        return self.__requests\n+\n     async def _queue_request(\n         self,\n         req: InferenceRequest,\n     ) -> Tuple[str, Awaitable[InferenceResponse]]:\n         internal_id = generate_uuid()\n+\n         await self._requests.put((internal_id, req))\n \n         loop = asyncio.get_running_loop()\ndiff --git a/mlserver/rest/errors.py b/mlserver/rest/errors.py\n--- a/mlserver/rest/errors.py\n+++ b/mlserver/rest/errors.py\n@@ -10,7 +10,7 @@\n     error: Optional[str] = None\n \n \n-def handle_mlserver_error(request: Request, exc: MLServerError) -> Response:\n+async def handle_mlserver_error(request: Request, exc: MLServerError) -> Response:\n     err_res = APIErrorResponse(error=str(exc))\n     return Response(status_code=status.HTTP_400_BAD_REQUEST, content=err_res.dict())\n \ndiff --git a/setup.py b/setup.py\n--- a/setup.py\n+++ b/setup.py\n@@ -35,9 +35,7 @@\n     packages=find_packages(exclude=[\"tests\", \"tests.*\"]),\n     install_requires=[\n         \"click\",\n-        # We pin version of fastapi\n-        # check https://github.com/SeldonIO/MLServer/issues/340\n-        \"fastapi==0.68.2\",\n+        \"fastapi\",\n         \"grpcio\",\n         \"importlib-metadata;python_version<'3.8'\",\n         \"numpy\",\n", "issue": "fastapi 0.70 failing tests\nWhen using `fastapi` latest version, some tests fail e.g. `test_infer`, if we use an older version of it (e.g. `0.68.2`) tests pass.\r\n\r\nWe need to take a deeper look just in case it uncovers a more subtle issue with asyncio (judging by the error we get).\r\n\r\nhere is the error for reference\r\n\r\n```\r\nest_infer[sum-model-None] _____________________________________\r\n\r\nrest_client = <starlette.testclient.TestClient object at 0x7fa456254bd0>\r\ninference_request = InferenceRequest(id=None, parameters=None, inputs=[RequestInput(name='input-0', shape=[1, 3], datatype='INT32', parameters=None, data=TensorData(__root__=[1, 2, 3]))], outputs=None)\r\nmodel_name = 'sum-model', model_version = None\r\n\r\n    @pytest.mark.parametrize(\r\n        \"model_name,model_version\", [(\"sum-model\", \"v1.2.3\"), (\"sum-model\", None)]\r\n    )\r\n    def test_infer(rest_client, inference_request, model_name, model_version):\r\n        endpoint = f\"/v2/models/{model_name}/infer\"\r\n        if model_version is not None:\r\n            endpoint = f\"/v2/models/{model_name}/versions/{model_version}/infer\"\r\n>       response = rest_client.post(endpoint, json=inference_request.dict())\r\n\r\ntests/rest/test_endpoints.py:57: \r\n_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _\r\n../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/requests/sessions.py:590: in post\r\n    return self.request('POST', url, data=data, json=json, **kwargs)\r\n../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/starlette/testclient.py:484: in request\r\n    json=json,\r\n../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/requests/sessions.py:542: in request\r\n    resp = self.send(prep, **send_kwargs)\r\n../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/requests/sessions.py:655: in send\r\n    r = adapter.send(request, **kwargs)\r\n../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/starlette/testclient.py:266: in send\r\n    raise exc\r\n../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/starlette/testclient.py:263: in send\r\n    portal.call(self.app, scope, receive, send)\r\n../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/anyio/from_thread.py:230: in call\r\n    return cast(T_Retval, self.start_task_soon(func, *args).result())\r\n../../miniconda3/envs/test_nest_io/lib/python3.7/concurrent/futures/_base.py:435: in result\r\n    return self.__get_result()\r\n../../miniconda3/envs/test_nest_io/lib/python3.7/concurrent/futures/_base.py:384: in __get_result\r\n    raise self._exception\r\n../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/anyio/from_thread.py:177: in _call_func\r\n    retval = await retval\r\n../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/fastapi/applications.py:208: in __call__\r\n    await super().__call__(scope, receive, send)\r\n../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/starlette/applications.py:112: in __call__\r\n    await self.middleware_stack(scope, receive, send)\r\n../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/starlette/middleware/errors.py:181: in __call__\r\n    raise exc\r\n../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/starlette/middleware/errors.py:159: in __call__\r\n    await self.app(scope, receive, _send)\r\n../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/starlette/exceptions.py:82: in __call__\r\n    raise exc\r\n../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/starlette/exceptions.py:71: in __call__\r\n    await self.app(scope, receive, sender)\r\n../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/starlette/routing.py:656: in __call__\r\n    await route.handle(scope, receive, send)\r\n../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/starlette/routing.py:259: in handle\r\n    await self.app(scope, receive, send)\r\n../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/starlette/routing.py:61: in app\r\n    response = await func(request)\r\nmlserver/rest/app.py:24: in custom_route_handler\r\n    return await original_route_handler(request)\r\n../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/fastapi/routing.py:227: in app\r\n    dependant=dependant, values=values, is_coroutine=is_coroutine\r\n../../miniconda3/envs/test_nest_io/lib/python3.7/site-packages/fastapi/routing.py:159: in run_endpoint_function\r\n    return await dependant.call(**values)\r\nmlserver/rest/endpoints.py:52: in infer\r\n    return await self._data_plane.infer(payload, model_name, model_version)\r\nmlserver/handlers/dataplane.py:60: in infer\r\n    prediction = await model.predict(payload)\r\nmlserver/batching/hooks.py:43: in _inner\r\n    return await batcher.predict(payload)\r\nmlserver/batching/adaptive.py:36: in predict\r\n    return await self._wait_response(internal_id)\r\nmlserver/batching/adaptive.py:55: in _wait_response\r\n    response = await async_response\r\nmlserver/batching/adaptive.py:85: in _batcher\r\n    async for batched in self._batch_requests():\r\nmlserver/batching/adaptive.py:112: in _batch_requests\r\n    read_op, timeout=timeout\r\n../../miniconda3/envs/test_nest_io/lib/python3.7/asyncio/tasks.py:442: in wait_for\r\n    return fut.result()\r\n_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _\r\n\r\nself = <Queue at 0x7fa456575490 maxsize=10 tasks=1>\r\n\r\n    async def get(self):\r\n        \"\"\"Remove and return an item from the queue.\r\n    \r\n        If queue is empty, wait until an item is available.\r\n        \"\"\"\r\n        while self.empty():\r\n            getter = self._loop.create_future()\r\n            self._getters.append(getter)\r\n            try:\r\n>               await getter\r\nE               RuntimeError: Task <Task pending coro=<Queue.get() running at /home/sa/miniconda3/envs/test_nest_io/lib/python3.7/asyncio/queues.py:159> cb=[_release_waiter(<Future pendi...4566d8550>()]>)() at /home/sa/miniconda3/envs/test_nest_io/lib/python3.7/asyncio/tasks.py:392]> got Future <Future pending> attached to a different loop\r\n\r\n../../miniconda3/envs/test_nest_io/lib/python3.7/asyncio/queues.py:159: RuntimeError\r\n```\r\n\n", "before_files": [{"content": "import time\nimport asyncio\n\nfrom asyncio import Future, Queue, wait_for, Task\nfrom functools import partial\nfrom typing import AsyncIterator, Awaitable, Dict, Tuple\n\nfrom ..model import MLModel\nfrom ..types import (\n    InferenceRequest,\n    InferenceResponse,\n)\nfrom ..utils import generate_uuid\n\nfrom .requests import BatchedRequests\n\n\nclass AdaptiveBatcher:\n    def __init__(self, model: MLModel):\n        self._model = model\n\n        self._max_batch_size = model.settings.max_batch_size\n        self._max_batch_time = model.settings.max_batch_time\n\n        # Save predict function before it gets decorated\n        self._predict_fn = model.predict\n        self._requests: Queue[Tuple[str, InferenceRequest]] = Queue(\n            maxsize=self._max_batch_size\n        )\n        self._async_responses: Dict[str, Future[InferenceResponse]] = {}\n        self._batching_task = None\n\n    async def predict(self, req: InferenceRequest) -> InferenceResponse:\n        internal_id, _ = await self._queue_request(req)\n        self._start_batcher_if_needed()\n        return await self._wait_response(internal_id)\n\n    async def _queue_request(\n        self,\n        req: InferenceRequest,\n    ) -> Tuple[str, Awaitable[InferenceResponse]]:\n        internal_id = generate_uuid()\n        await self._requests.put((internal_id, req))\n\n        loop = asyncio.get_running_loop()\n        async_response = loop.create_future()\n        self._async_responses[internal_id] = async_response\n\n        return internal_id, async_response\n\n    async def _wait_response(self, internal_id: str) -> InferenceResponse:\n        async_response = self._async_responses[internal_id]\n\n        try:\n            response = await async_response\n            return response\n        finally:\n            del self._async_responses[internal_id]\n\n    def _start_batcher_if_needed(self):\n        if self._batching_task is not None:\n            if not self._batching_task.done():\n                # If task hasn't finished yet, let it keep running\n                return\n\n        self._batching_task = asyncio.create_task(self._batcher())\n        self._batching_task.add_done_callback(self._batching_task_callback)\n\n    def _batching_task_callback(self, batching_task: Task):\n        err = batching_task.exception()\n        if err:\n            # Clear queue\n            self._clear_queue(err)\n\n    def _clear_queue(self, err: BaseException):\n        # Cancel all pending async responses\n        for async_response in self._async_responses.values():\n            async_response.set_exception(err)\n\n        # Empty queue\n        for _ in range(self._requests.qsize()):\n            self._requests.get_nowait()\n\n    async def _batcher(self):\n        async for batched in self._batch_requests():\n            # We run prediction as a Task to ensure it gets scheduled\n            # immediately.\n            # That way, we can process multiple batches concurrently.\n            predict_task = asyncio.create_task(self._predict_fn(batched.merged_request))\n            predict_task.add_done_callback(partial(self._predict_callback, batched))\n\n    def _predict_callback(self, batched: BatchedRequests, predict_task: Task):\n        try:\n            batched_response = predict_task.result()\n            responses = batched.split_response(batched_response)\n            for internal_id, response in responses.items():\n                self._async_responses[internal_id].set_result(response)\n        except Exception as err:\n            for internal_id in batched.inference_requests.keys():\n                self._async_responses[internal_id].set_exception(err)\n\n    async def _batch_requests(self) -> AsyncIterator[BatchedRequests]:\n        while not self._requests.empty():\n            to_batch: Dict[str, InferenceRequest] = {}\n            start = time.time()\n            timeout = self._max_batch_time\n\n            try:\n                while len(to_batch) < self._max_batch_size:\n                    read_op = self._requests.get()\n                    internal_id, inference_request = await wait_for(\n                        read_op, timeout=timeout\n                    )\n                    to_batch[internal_id] = inference_request\n\n                    # Update remaining timeout\n                    current = time.time()\n                    timeout = timeout - (current - start)\n            except asyncio.TimeoutError:\n                # NOTE: Hit timeout, continue\n                pass\n\n            yield BatchedRequests(to_batch)\n", "path": "mlserver/batching/adaptive.py"}, {"content": "import os\n\nfrom typing import Dict\nfrom setuptools import setup, find_packages\n\nROOT_PATH = os.path.dirname(__file__)\nPKG_NAME = \"mlserver\"\nPKG_PATH = os.path.join(ROOT_PATH, PKG_NAME)\n\n\ndef _load_version() -> str:\n    version = \"\"\n    version_path = os.path.join(PKG_PATH, \"version.py\")\n    with open(version_path) as fp:\n        version_module: Dict[str, str] = {}\n        exec(fp.read(), version_module)\n        version = version_module[\"__version__\"]\n\n    return version\n\n\ndef _load_description() -> str:\n    readme_path = os.path.join(ROOT_PATH, \"README.md\")\n    with open(readme_path) as fp:\n        return fp.read()\n\n\nsetup(\n    name=PKG_NAME,\n    version=_load_version(),\n    url=\"https://github.com/SeldonIO/MLServer.git\",\n    author=\"Seldon Technologies Ltd.\",\n    author_email=\"hello@seldon.io\",\n    description=\"ML server\",\n    packages=find_packages(exclude=[\"tests\", \"tests.*\"]),\n    install_requires=[\n        \"click\",\n        # We pin version of fastapi\n        # check https://github.com/SeldonIO/MLServer/issues/340\n        \"fastapi==0.68.2\",\n        \"grpcio\",\n        \"importlib-metadata;python_version<'3.8'\",\n        \"numpy\",\n        \"pandas\",\n        \"protobuf\",\n        \"uvicorn\",\n        \"starlette_exporter\",\n        \"py-grpc-prometheus\",\n    ],\n    extras_require={\"all\": [\"orjson\"]},\n    entry_points={\"console_scripts\": [\"mlserver=mlserver.cli:main\"]},\n    long_description=_load_description(),\n    long_description_content_type=\"text/markdown\",\n    license=\"Apache 2.0\",\n)\n", "path": "setup.py"}, {"content": "from typing import Optional\nfrom fastapi import Request, status\nfrom pydantic import BaseModel\n\nfrom .responses import Response\nfrom ..errors import MLServerError\n\n\nclass APIErrorResponse(BaseModel):\n    error: Optional[str] = None\n\n\ndef handle_mlserver_error(request: Request, exc: MLServerError) -> Response:\n    err_res = APIErrorResponse(error=str(exc))\n    return Response(status_code=status.HTTP_400_BAD_REQUEST, content=err_res.dict())\n\n\n_EXCEPTION_HANDLERS = {MLServerError: handle_mlserver_error}\n", "path": "mlserver/rest/errors.py"}], "after_files": [{"content": "# Required to deal with annotations including `Queue`\n# https://mypy.readthedocs.io/en/latest/common_issues.html#issues-with-code-at-runtime\nfrom __future__ import annotations\n\nimport time\nimport asyncio\n\nfrom asyncio import Future, Queue, wait_for, Task\nfrom functools import partial\nfrom typing import AsyncIterator, Awaitable, Dict, Optional, Tuple\n\nfrom ..model import MLModel\nfrom ..types import (\n    InferenceRequest,\n    InferenceResponse,\n)\nfrom ..utils import generate_uuid\n\nfrom .requests import BatchedRequests\n\n\nclass AdaptiveBatcher:\n    def __init__(self, model: MLModel):\n        self._model = model\n\n        self._max_batch_size = model.settings.max_batch_size\n        self._max_batch_time = model.settings.max_batch_time\n\n        # Save predict function before it gets decorated\n        self._predict_fn = model.predict\n        self.__requests: Optional[Queue[Tuple[str, InferenceRequest]]] = None\n        self._async_responses: Dict[str, Future[InferenceResponse]] = {}\n        self._batching_task = None\n\n    async def predict(self, req: InferenceRequest) -> InferenceResponse:\n        internal_id, _ = await self._queue_request(req)\n        self._start_batcher_if_needed()\n        return await self._wait_response(internal_id)\n\n    @property\n    def _requests(self) -> Queue[Tuple[str, InferenceRequest]]:\n        # NOTE: We need to create Queue within the async request path (and not\n        # during __init__!!) to ensure that it shares the same AsyncIO loop.\n        if self.__requests is None:\n            self.__requests = Queue(maxsize=self._max_batch_size)\n\n        return self.__requests\n\n    async def _queue_request(\n        self,\n        req: InferenceRequest,\n    ) -> Tuple[str, Awaitable[InferenceResponse]]:\n        internal_id = generate_uuid()\n\n        await self._requests.put((internal_id, req))\n\n        loop = asyncio.get_running_loop()\n        async_response = loop.create_future()\n        self._async_responses[internal_id] = async_response\n\n        return internal_id, async_response\n\n    async def _wait_response(self, internal_id: str) -> InferenceResponse:\n        async_response = self._async_responses[internal_id]\n\n        try:\n            response = await async_response\n            return response\n        finally:\n            del self._async_responses[internal_id]\n\n    def _start_batcher_if_needed(self):\n        if self._batching_task is not None:\n            if not self._batching_task.done():\n                # If task hasn't finished yet, let it keep running\n                return\n\n        self._batching_task = asyncio.create_task(self._batcher())\n        self._batching_task.add_done_callback(self._batching_task_callback)\n\n    def _batching_task_callback(self, batching_task: Task):\n        err = batching_task.exception()\n        if err:\n            # Clear queue\n            self._clear_queue(err)\n\n    def _clear_queue(self, err: BaseException):\n        # Cancel all pending async responses\n        for async_response in self._async_responses.values():\n            async_response.set_exception(err)\n\n        # Empty queue\n        for _ in range(self._requests.qsize()):\n            self._requests.get_nowait()\n\n    async def _batcher(self):\n        async for batched in self._batch_requests():\n            # We run prediction as a Task to ensure it gets scheduled\n            # immediately.\n            # That way, we can process multiple batches concurrently.\n            predict_task = asyncio.create_task(self._predict_fn(batched.merged_request))\n            predict_task.add_done_callback(partial(self._predict_callback, batched))\n\n    def _predict_callback(self, batched: BatchedRequests, predict_task: Task):\n        try:\n            batched_response = predict_task.result()\n            responses = batched.split_response(batched_response)\n            for internal_id, response in responses.items():\n                self._async_responses[internal_id].set_result(response)\n        except Exception as err:\n            for internal_id in batched.inference_requests.keys():\n                self._async_responses[internal_id].set_exception(err)\n\n    async def _batch_requests(self) -> AsyncIterator[BatchedRequests]:\n        while not self._requests.empty():\n            to_batch: Dict[str, InferenceRequest] = {}\n            start = time.time()\n            timeout = self._max_batch_time\n\n            try:\n                while len(to_batch) < self._max_batch_size:\n                    read_op = self._requests.get()\n                    internal_id, inference_request = await wait_for(\n                        read_op, timeout=timeout\n                    )\n                    to_batch[internal_id] = inference_request\n\n                    # Update remaining timeout\n                    current = time.time()\n                    timeout = timeout - (current - start)\n            except asyncio.TimeoutError:\n                # NOTE: Hit timeout, continue\n                pass\n\n            yield BatchedRequests(to_batch)\n", "path": "mlserver/batching/adaptive.py"}, {"content": "import os\n\nfrom typing import Dict\nfrom setuptools import setup, find_packages\n\nROOT_PATH = os.path.dirname(__file__)\nPKG_NAME = \"mlserver\"\nPKG_PATH = os.path.join(ROOT_PATH, PKG_NAME)\n\n\ndef _load_version() -> str:\n    version = \"\"\n    version_path = os.path.join(PKG_PATH, \"version.py\")\n    with open(version_path) as fp:\n        version_module: Dict[str, str] = {}\n        exec(fp.read(), version_module)\n        version = version_module[\"__version__\"]\n\n    return version\n\n\ndef _load_description() -> str:\n    readme_path = os.path.join(ROOT_PATH, \"README.md\")\n    with open(readme_path) as fp:\n        return fp.read()\n\n\nsetup(\n    name=PKG_NAME,\n    version=_load_version(),\n    url=\"https://github.com/SeldonIO/MLServer.git\",\n    author=\"Seldon Technologies Ltd.\",\n    author_email=\"hello@seldon.io\",\n    description=\"ML server\",\n    packages=find_packages(exclude=[\"tests\", \"tests.*\"]),\n    install_requires=[\n        \"click\",\n        \"fastapi\",\n        \"grpcio\",\n        \"importlib-metadata;python_version<'3.8'\",\n        \"numpy\",\n        \"pandas\",\n        \"protobuf\",\n        \"uvicorn\",\n        \"starlette_exporter\",\n        \"py-grpc-prometheus\",\n    ],\n    extras_require={\"all\": [\"orjson\"]},\n    entry_points={\"console_scripts\": [\"mlserver=mlserver.cli:main\"]},\n    long_description=_load_description(),\n    long_description_content_type=\"text/markdown\",\n    license=\"Apache 2.0\",\n)\n", "path": "setup.py"}, {"content": "from typing import Optional\nfrom fastapi import Request, status\nfrom pydantic import BaseModel\n\nfrom .responses import Response\nfrom ..errors import MLServerError\n\n\nclass APIErrorResponse(BaseModel):\n    error: Optional[str] = None\n\n\nasync def handle_mlserver_error(request: Request, exc: MLServerError) -> Response:\n    err_res = APIErrorResponse(error=str(exc))\n    return Response(status_code=status.HTTP_400_BAD_REQUEST, content=err_res.dict())\n\n\n_EXCEPTION_HANDLERS = {MLServerError: handle_mlserver_error}\n", "path": "mlserver/rest/errors.py"}]}