We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
500 Error on MacOS
I'm running MLServer on MacOS (arm). I was following [Serving XGBoost models](https://mlserver.readthedocs.io/en/latest/examples/xgboost/README.html) example. I receive a 500 error when requesting a model with url `http://localhost:8080/v2/models/mushroom-xgboost/versions/v0.1.0/infer`.

MLServer throws an error

```
...
  File "/opt/homebrew/Cellar/python@3.10/3.10.9/Frameworks/Python.framework/Versions/3.10/lib/python3.10/multiprocessing/queues.py", line 126, in qsize
    return self._maxsize - self._sem._semlock._get_value()
NotImplementedError
```

Developers of `queues.py` left a comment that this is broken on MacOS 😬 

```python
    def qsize(self):
        # Raises NotImplementedError on Mac OSX because of broken sem_getvalue()
        return self._maxsize - self._sem._semlock._get_value()
```

Is it possible to get around this when running models with MLServer?
Thanks!

--- 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/parallel/dispatcher.py`
Content:
```
1 import asyncio
2 
3 from typing import Dict, List, Tuple
4 from itertools import cycle
5 from multiprocessing import Queue
6 from concurrent.futures import ThreadPoolExecutor
7 from asyncio import Future
8 
9 from ..utils import schedule_with_callback, generate_uuid
10 from ..metrics import REGISTRY
11 
12 from .worker import Worker
13 from .logging import logger
14 from .utils import END_OF_QUEUE, cancel_task
15 from .messages import (
16     Message,
17     ModelUpdateMessage,
18     ModelRequestMessage,
19     ModelResponseMessage,
20 )
21 from prometheus_client import Histogram
22 
23 QUEUE_METRIC_NAME = "parallel_request_queue"
24 
25 
26 class Dispatcher:
27     def __init__(self, workers: Dict[int, Worker], responses: Queue):
28         self._responses = responses
29         self._workers = workers
30         self._workers_round_robin = cycle(self._workers.keys())
31         self._active = False
32         self._process_responses_task = None
33         self._executor = ThreadPoolExecutor()
34         self._async_responses: Dict[str, Future[ModelResponseMessage]] = {}
35         self.parallel_request_queue_size = self._get_or_create_metric()
36 
37     def _get_or_create_metric(self) -> Histogram:
38         if QUEUE_METRIC_NAME in REGISTRY:
39             return REGISTRY[QUEUE_METRIC_NAME]  # type: ignore
40 
41         return Histogram(
42             QUEUE_METRIC_NAME,
43             "counter of request queue size for workers",
44             ["workerpid"],
45             registry=REGISTRY,
46         )
47 
48     def start(self):
49         self._active = True
50         self._process_responses_task = schedule_with_callback(
51             self._process_responses(), self._process_responses_cb
52         )
53 
54     def _process_responses_cb(self, process_responses):
55         try:
56             process_responses.result()
57         except asyncio.CancelledError:
58             # NOTE: The response loop was cancelled from the outside, so don't
59             # restart
60             return
61         except Exception:
62             logger.exception("Response processing loop crashed. Restarting the loop...")
63             # If process loop crashed, restart it
64             self.start()
65 
66     async def _process_responses(self):
67         logger.debug("Starting response processing loop...")
68         loop = asyncio.get_event_loop()
69         while self._active:
70             response = await loop.run_in_executor(self._executor, self._responses.get)
71 
72             # If the queue gets terminated, detect the "sentinel value" and
73             # stop reading
74             if response is END_OF_QUEUE:
75                 return
76 
77             await self._process_response(response)
78 
79     async def _process_response(self, response: ModelResponseMessage):
80         internal_id = response.id
81 
82         async_response = self._async_responses[internal_id]
83 
84         # NOTE: Use call_soon_threadsafe to cover cases where `model.predict()`
85         # (or other methods) get called from a separate thread (and a separate
86         # AsyncIO loop)
87         response_loop = async_response.get_loop()
88         if response.exception:
89             response_loop.call_soon_threadsafe(
90                 async_response.set_exception, response.exception
91             )
92         else:
93             response_loop.call_soon_threadsafe(async_response.set_result, response)
94 
95     async def dispatch_request(
96         self, request_message: ModelRequestMessage
97     ) -> ModelResponseMessage:
98         worker, wpid = self._get_worker()
99         self._workers_queue_monitor(worker, wpid)
100         worker.send_request(request_message)
101 
102         return await self._dispatch(request_message)
103 
104     def _get_worker(self) -> Tuple[Worker, int]:
105         """
106         Get next available worker.
107         By default, this is just a round-robin through all the workers.
108         """
109         worker_pid = next(self._workers_round_robin)
110         return self._workers[worker_pid], worker_pid
111 
112     def _workers_queue_monitor(self, worker: Worker, worker_pid: int):
113         """Get metrics from every worker request queue"""
114         queue_size = worker._requests.qsize()
115 
116         self.parallel_request_queue_size.labels(workerpid=str(worker_pid)).observe(
117             float(queue_size)
118         )
119 
120     async def dispatch_update(
121         self, model_update: ModelUpdateMessage
122     ) -> List[ModelResponseMessage]:
123         return await asyncio.gather(
124             *[
125                 self._dispatch_update(worker, model_update)
126                 for worker in self._workers.values()
127             ]
128         )
129 
130     async def _dispatch_update(
131         self, worker: Worker, model_update: ModelUpdateMessage
132     ) -> ModelResponseMessage:
133         # NOTE: Need to rewrite the UUID to ensure each worker sends back a
134         # unique result
135         worker_update = model_update.copy()
136         worker_update.id = generate_uuid()
137         worker.send_update(worker_update)
138         return await self._dispatch(worker_update)
139 
140     async def _dispatch(self, message: Message) -> ModelResponseMessage:
141         loop = asyncio.get_running_loop()
142         async_response = loop.create_future()
143         internal_id = message.id
144         self._async_responses[internal_id] = async_response
145 
146         return await self._wait_response(internal_id)
147 
148     async def _wait_response(self, internal_id: str) -> ModelResponseMessage:
149         async_response = self._async_responses[internal_id]
150 
151         try:
152             inference_response = await async_response
153             return inference_response
154         finally:
155             del self._async_responses[internal_id]
156 
157     async def stop(self):
158         self._executor.shutdown()
159         if self._process_responses_task is not None:
160             await cancel_task(self._process_responses_task)
161
```
--- 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/parallel/dispatcher.py b/mlserver/parallel/dispatcher.py
--- a/mlserver/parallel/dispatcher.py
+++ b/mlserver/parallel/dispatcher.py
@@ -41,7 +41,6 @@
         return Histogram(
             QUEUE_METRIC_NAME,
             "counter of request queue size for workers",
-            ["workerpid"],
             registry=REGISTRY,
         )
 
@@ -96,7 +95,6 @@
         self, request_message: ModelRequestMessage
     ) -> ModelResponseMessage:
         worker, wpid = self._get_worker()
-        self._workers_queue_monitor(worker, wpid)
         worker.send_request(request_message)
 
         return await self._dispatch(request_message)
@@ -109,14 +107,6 @@
         worker_pid = next(self._workers_round_robin)
         return self._workers[worker_pid], worker_pid
 
-    def _workers_queue_monitor(self, worker: Worker, worker_pid: int):
-        """Get metrics from every worker request queue"""
-        queue_size = worker._requests.qsize()
-
-        self.parallel_request_queue_size.labels(workerpid=str(worker_pid)).observe(
-            float(queue_size)
-        )
-
     async def dispatch_update(
         self, model_update: ModelUpdateMessage
     ) -> List[ModelResponseMessage]:
@@ -143,6 +133,8 @@
         internal_id = message.id
         self._async_responses[internal_id] = async_response
 
+        # Monitor current in-flight requests
+        self.parallel_request_queue_size.observe(len(self._async_responses))
         return await self._wait_response(internal_id)
 
     async def _wait_response(self, internal_id: str) -> ModelResponseMessage:

{"golden_diff": "diff --git a/mlserver/parallel/dispatcher.py b/mlserver/parallel/dispatcher.py\n--- a/mlserver/parallel/dispatcher.py\n+++ b/mlserver/parallel/dispatcher.py\n@@ -41,7 +41,6 @@\n         return Histogram(\n             QUEUE_METRIC_NAME,\n             \"counter of request queue size for workers\",\n-            [\"workerpid\"],\n             registry=REGISTRY,\n         )\n \n@@ -96,7 +95,6 @@\n         self, request_message: ModelRequestMessage\n     ) -> ModelResponseMessage:\n         worker, wpid = self._get_worker()\n-        self._workers_queue_monitor(worker, wpid)\n         worker.send_request(request_message)\n \n         return await self._dispatch(request_message)\n@@ -109,14 +107,6 @@\n         worker_pid = next(self._workers_round_robin)\n         return self._workers[worker_pid], worker_pid\n \n-    def _workers_queue_monitor(self, worker: Worker, worker_pid: int):\n-        \"\"\"Get metrics from every worker request queue\"\"\"\n-        queue_size = worker._requests.qsize()\n-\n-        self.parallel_request_queue_size.labels(workerpid=str(worker_pid)).observe(\n-            float(queue_size)\n-        )\n-\n     async def dispatch_update(\n         self, model_update: ModelUpdateMessage\n     ) -> List[ModelResponseMessage]:\n@@ -143,6 +133,8 @@\n         internal_id = message.id\n         self._async_responses[internal_id] = async_response\n \n+        # Monitor current in-flight requests\n+        self.parallel_request_queue_size.observe(len(self._async_responses))\n         return await self._wait_response(internal_id)\n \n     async def _wait_response(self, internal_id: str) -> ModelResponseMessage:\n", "issue": "500 Error on MacOS\nI'm running MLServer on MacOS (arm). I was following [Serving XGBoost models](https://mlserver.readthedocs.io/en/latest/examples/xgboost/README.html) example. I receive a 500 error when requesting a model with url `http://localhost:8080/v2/models/mushroom-xgboost/versions/v0.1.0/infer`.\r\n\r\nMLServer throws an error\r\n\r\n```\r\n...\r\n  File \"/opt/homebrew/Cellar/python@3.10/3.10.9/Frameworks/Python.framework/Versions/3.10/lib/python3.10/multiprocessing/queues.py\", line 126, in qsize\r\n    return self._maxsize - self._sem._semlock._get_value()\r\nNotImplementedError\r\n```\r\n\r\nDevelopers of `queues.py` left a comment that this is broken on MacOS \ud83d\ude2c \r\n\r\n```python\r\n    def qsize(self):\r\n        # Raises NotImplementedError on Mac OSX because of broken sem_getvalue()\r\n        return self._maxsize - self._sem._semlock._get_value()\r\n```\r\n\r\nIs it possible to get around this when running models with MLServer?\r\nThanks!\n", "before_files": [{"content": "import asyncio\n\nfrom typing import Dict, List, Tuple\nfrom itertools import cycle\nfrom multiprocessing import Queue\nfrom concurrent.futures import ThreadPoolExecutor\nfrom asyncio import Future\n\nfrom ..utils import schedule_with_callback, generate_uuid\nfrom ..metrics import REGISTRY\n\nfrom .worker import Worker\nfrom .logging import logger\nfrom .utils import END_OF_QUEUE, cancel_task\nfrom .messages import (\n    Message,\n    ModelUpdateMessage,\n    ModelRequestMessage,\n    ModelResponseMessage,\n)\nfrom prometheus_client import Histogram\n\nQUEUE_METRIC_NAME = \"parallel_request_queue\"\n\n\nclass Dispatcher:\n    def __init__(self, workers: Dict[int, Worker], responses: Queue):\n        self._responses = responses\n        self._workers = workers\n        self._workers_round_robin = cycle(self._workers.keys())\n        self._active = False\n        self._process_responses_task = None\n        self._executor = ThreadPoolExecutor()\n        self._async_responses: Dict[str, Future[ModelResponseMessage]] = {}\n        self.parallel_request_queue_size = self._get_or_create_metric()\n\n    def _get_or_create_metric(self) -> Histogram:\n        if QUEUE_METRIC_NAME in REGISTRY:\n            return REGISTRY[QUEUE_METRIC_NAME]  # type: ignore\n\n        return Histogram(\n            QUEUE_METRIC_NAME,\n            \"counter of request queue size for workers\",\n            [\"workerpid\"],\n            registry=REGISTRY,\n        )\n\n    def start(self):\n        self._active = True\n        self._process_responses_task = schedule_with_callback(\n            self._process_responses(), self._process_responses_cb\n        )\n\n    def _process_responses_cb(self, process_responses):\n        try:\n            process_responses.result()\n        except asyncio.CancelledError:\n            # NOTE: The response loop was cancelled from the outside, so don't\n            # restart\n            return\n        except Exception:\n            logger.exception(\"Response processing loop crashed. Restarting the loop...\")\n            # If process loop crashed, restart it\n            self.start()\n\n    async def _process_responses(self):\n        logger.debug(\"Starting response processing loop...\")\n        loop = asyncio.get_event_loop()\n        while self._active:\n            response = await loop.run_in_executor(self._executor, self._responses.get)\n\n            # If the queue gets terminated, detect the \"sentinel value\" and\n            # stop reading\n            if response is END_OF_QUEUE:\n                return\n\n            await self._process_response(response)\n\n    async def _process_response(self, response: ModelResponseMessage):\n        internal_id = response.id\n\n        async_response = self._async_responses[internal_id]\n\n        # NOTE: Use call_soon_threadsafe to cover cases where `model.predict()`\n        # (or other methods) get called from a separate thread (and a separate\n        # AsyncIO loop)\n        response_loop = async_response.get_loop()\n        if response.exception:\n            response_loop.call_soon_threadsafe(\n                async_response.set_exception, response.exception\n            )\n        else:\n            response_loop.call_soon_threadsafe(async_response.set_result, response)\n\n    async def dispatch_request(\n        self, request_message: ModelRequestMessage\n    ) -> ModelResponseMessage:\n        worker, wpid = self._get_worker()\n        self._workers_queue_monitor(worker, wpid)\n        worker.send_request(request_message)\n\n        return await self._dispatch(request_message)\n\n    def _get_worker(self) -> Tuple[Worker, int]:\n        \"\"\"\n        Get next available worker.\n        By default, this is just a round-robin through all the workers.\n        \"\"\"\n        worker_pid = next(self._workers_round_robin)\n        return self._workers[worker_pid], worker_pid\n\n    def _workers_queue_monitor(self, worker: Worker, worker_pid: int):\n        \"\"\"Get metrics from every worker request queue\"\"\"\n        queue_size = worker._requests.qsize()\n\n        self.parallel_request_queue_size.labels(workerpid=str(worker_pid)).observe(\n            float(queue_size)\n        )\n\n    async def dispatch_update(\n        self, model_update: ModelUpdateMessage\n    ) -> List[ModelResponseMessage]:\n        return await asyncio.gather(\n            *[\n                self._dispatch_update(worker, model_update)\n                for worker in self._workers.values()\n            ]\n        )\n\n    async def _dispatch_update(\n        self, worker: Worker, model_update: ModelUpdateMessage\n    ) -> ModelResponseMessage:\n        # NOTE: Need to rewrite the UUID to ensure each worker sends back a\n        # unique result\n        worker_update = model_update.copy()\n        worker_update.id = generate_uuid()\n        worker.send_update(worker_update)\n        return await self._dispatch(worker_update)\n\n    async def _dispatch(self, message: Message) -> ModelResponseMessage:\n        loop = asyncio.get_running_loop()\n        async_response = loop.create_future()\n        internal_id = message.id\n        self._async_responses[internal_id] = async_response\n\n        return await self._wait_response(internal_id)\n\n    async def _wait_response(self, internal_id: str) -> ModelResponseMessage:\n        async_response = self._async_responses[internal_id]\n\n        try:\n            inference_response = await async_response\n            return inference_response\n        finally:\n            del self._async_responses[internal_id]\n\n    async def stop(self):\n        self._executor.shutdown()\n        if self._process_responses_task is not None:\n            await cancel_task(self._process_responses_task)\n", "path": "mlserver/parallel/dispatcher.py"}], "after_files": [{"content": "import asyncio\n\nfrom typing import Dict, List, Tuple\nfrom itertools import cycle\nfrom multiprocessing import Queue\nfrom concurrent.futures import ThreadPoolExecutor\nfrom asyncio import Future\n\nfrom ..utils import schedule_with_callback, generate_uuid\nfrom ..metrics import REGISTRY\n\nfrom .worker import Worker\nfrom .logging import logger\nfrom .utils import END_OF_QUEUE, cancel_task\nfrom .messages import (\n    Message,\n    ModelUpdateMessage,\n    ModelRequestMessage,\n    ModelResponseMessage,\n)\nfrom prometheus_client import Histogram\n\nQUEUE_METRIC_NAME = \"parallel_request_queue\"\n\n\nclass Dispatcher:\n    def __init__(self, workers: Dict[int, Worker], responses: Queue):\n        self._responses = responses\n        self._workers = workers\n        self._workers_round_robin = cycle(self._workers.keys())\n        self._active = False\n        self._process_responses_task = None\n        self._executor = ThreadPoolExecutor()\n        self._async_responses: Dict[str, Future[ModelResponseMessage]] = {}\n        self.parallel_request_queue_size = self._get_or_create_metric()\n\n    def _get_or_create_metric(self) -> Histogram:\n        if QUEUE_METRIC_NAME in REGISTRY:\n            return REGISTRY[QUEUE_METRIC_NAME]  # type: ignore\n\n        return Histogram(\n            QUEUE_METRIC_NAME,\n            \"counter of request queue size for workers\",\n            registry=REGISTRY,\n        )\n\n    def start(self):\n        self._active = True\n        self._process_responses_task = schedule_with_callback(\n            self._process_responses(), self._process_responses_cb\n        )\n\n    def _process_responses_cb(self, process_responses):\n        try:\n            process_responses.result()\n        except asyncio.CancelledError:\n            # NOTE: The response loop was cancelled from the outside, so don't\n            # restart\n            return\n        except Exception:\n            logger.exception(\"Response processing loop crashed. Restarting the loop...\")\n            # If process loop crashed, restart it\n            self.start()\n\n    async def _process_responses(self):\n        logger.debug(\"Starting response processing loop...\")\n        loop = asyncio.get_event_loop()\n        while self._active:\n            response = await loop.run_in_executor(self._executor, self._responses.get)\n\n            # If the queue gets terminated, detect the \"sentinel value\" and\n            # stop reading\n            if response is END_OF_QUEUE:\n                return\n\n            await self._process_response(response)\n\n    async def _process_response(self, response: ModelResponseMessage):\n        internal_id = response.id\n\n        async_response = self._async_responses[internal_id]\n\n        # NOTE: Use call_soon_threadsafe to cover cases where `model.predict()`\n        # (or other methods) get called from a separate thread (and a separate\n        # AsyncIO loop)\n        response_loop = async_response.get_loop()\n        if response.exception:\n            response_loop.call_soon_threadsafe(\n                async_response.set_exception, response.exception\n            )\n        else:\n            response_loop.call_soon_threadsafe(async_response.set_result, response)\n\n    async def dispatch_request(\n        self, request_message: ModelRequestMessage\n    ) -> ModelResponseMessage:\n        worker, wpid = self._get_worker()\n        worker.send_request(request_message)\n\n        return await self._dispatch(request_message)\n\n    def _get_worker(self) -> Tuple[Worker, int]:\n        \"\"\"\n        Get next available worker.\n        By default, this is just a round-robin through all the workers.\n        \"\"\"\n        worker_pid = next(self._workers_round_robin)\n        return self._workers[worker_pid], worker_pid\n\n    async def dispatch_update(\n        self, model_update: ModelUpdateMessage\n    ) -> List[ModelResponseMessage]:\n        return await asyncio.gather(\n            *[\n                self._dispatch_update(worker, model_update)\n                for worker in self._workers.values()\n            ]\n        )\n\n    async def _dispatch_update(\n        self, worker: Worker, model_update: ModelUpdateMessage\n    ) -> ModelResponseMessage:\n        # NOTE: Need to rewrite the UUID to ensure each worker sends back a\n        # unique result\n        worker_update = model_update.copy()\n        worker_update.id = generate_uuid()\n        worker.send_update(worker_update)\n        return await self._dispatch(worker_update)\n\n    async def _dispatch(self, message: Message) -> ModelResponseMessage:\n        loop = asyncio.get_running_loop()\n        async_response = loop.create_future()\n        internal_id = message.id\n        self._async_responses[internal_id] = async_response\n\n        # Monitor current in-flight requests\n        self.parallel_request_queue_size.observe(len(self._async_responses))\n        return await self._wait_response(internal_id)\n\n    async def _wait_response(self, internal_id: str) -> ModelResponseMessage:\n        async_response = self._async_responses[internal_id]\n\n        try:\n            inference_response = await async_response\n            return inference_response\n        finally:\n            del self._async_responses[internal_id]\n\n    async def stop(self):\n        self._executor.shutdown()\n        if self._process_responses_task is not None:\n            await cancel_task(self._process_responses_task)\n", "path": "mlserver/parallel/dispatcher.py"}]}