Don't use qsize for parallel worker queue

.github/workflows/tests.yml

@@ -25,9 +25,10 @@ jobs:
           make lint
 
   tests:
-    runs-on: ubuntu-latest
     strategy:
+      fail-fast: false
       matrix:
+        os: [ubuntu-latest, macos-latest]
         python-version: ["3.8", "3.9", "3.10"]
         tox-environment:
           - mlserver
@@ -38,7 +39,18 @@ jobs:
           - huggingface
           - alibi-explain
           - alibi-detect
+    runs-on: ${{ matrix.os }}
     steps:
+      - name: Setup docker (missing on MacOS)
+        if: runner.os == 'macos'
+        run: |
+          # From https://github.com/actions/runner-images/issues/17#issuecomment-1537238473
+          # From https://github.com/abiosoft/colima/discussions/273#discussioncomment-4959736
+          brew install docker docker-buildx
+          mkdir -p $HOME/.docker/cli-plugins
+          ln -sfn $(which docker-buildx) $HOME/.docker/cli-plugins/docker-buildx
+          colima start --memory 4
+          sudo ln -sf $HOME/.colima/default/docker.sock /var/run/docker.sock
       - uses: actions/checkout@v3
       - name: Set up Python ${{ matrix.python-version }}
         uses: actions/setup-python@v2
@@ -56,12 +68,24 @@ jobs:
           tox -e ${{ matrix.tox-environment }}
 
   all-runtimes:
-    runs-on: ubuntu-latest
     if: github.event_name == 'push'
     strategy:
+      fail-fast: false
       matrix:
+        os: [ubuntu-latest, macos-latest]
         python-version: ["3.8", "3.9", "3.10"]
+    runs-on: ${{ matrix.os }}
     steps:
+      - name: Setup docker (missing on MacOS)
+        if: runner.os == 'macos'
+        run: |
+          # From https://github.com/actions/runner-images/issues/17#issuecomment-1537238473
+          # From https://github.com/abiosoft/colima/discussions/273#discussioncomment-4959736
+          brew install docker docker-buildx
+          mkdir -p $HOME/.docker/cli-plugins
+          ln -sfn $(which docker-buildx) $HOME/.docker/cli-plugins/docker-buildx
+          colima start --memory 4
+          sudo ln -sf $HOME/.colima/default/docker.sock /var/run/docker.sock
       - uses: actions/checkout@v3
       - name: Set up Python ${{ matrix.python-version }}
         uses: actions/setup-python@v2

mlserver/parallel/dispatcher.py

@@ -41,7 +41,6 @@ def _get_or_create_metric(self) -> Histogram:
         return Histogram(
             QUEUE_METRIC_NAME,
             "counter of request queue size for workers",
-            ["workerpid"],
             registry=REGISTRY,
         )
 
@@ -96,7 +95,6 @@ async def dispatch_request(
         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 @@ def _get_worker(self) -> Tuple[Worker, int]:
         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 @@ async def _dispatch(self, message: Message) -> ModelResponseMessage:
         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:

tests/utils.py

@@ -120,7 +120,12 @@ async def _retry_get(self, endpoint: str):
             attempts=10,
             start_timeout=0.5,
             statuses=[400],
-            exceptions={ClientConnectorError, ClientOSError, ServerDisconnectedError},
+            exceptions={
+                ClientConnectorError,
+                ClientOSError,
+                ServerDisconnectedError,
+                ConnectionRefusedError,
+            },
         )
         retry_client = RetryClient(raise_for_status=True, retry_options=retry_options)