prompt: 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()
  
  ```
  
golden_diff: 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:
  
verification_info: {"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"}]}