code: def test_long_poll_timeout_with_max_concurrent_queries(ray_instance):
      """Test that max_concurrent_queries is respected when there are long poll timeouts.
  
      Previously, when a long poll update occurred (e.g., a timeout or new replicas
      added), ongoing requests would no longer be counted against
      `max_concurrent_queries`.
  
      Issue: https://github.com/ray-project/ray/issues/32652
      """
  
      @ray.remote(num_cpus=0)
      class CounterActor:
          def __init__(self):
              self._count = 0
  
          def inc(self):
              self._count += 1
  
          def get(self):
              return self._count
  
      signal_actor = SignalActor.remote()
      counter_actor = CounterActor.remote()
  
      @serve.deployment(max_concurrent_queries=1)
      async def f():
          await counter_actor.inc.remote()
          await signal_actor.wait.remote()
          return "hello"
  
      # Issue a blocking request which should occupy the only slot due to
      # `max_concurrent_queries=1`.
      serve.run(f.bind())
  
      @ray.remote
      def do_req():
          return requests.get("http://localhost:8000").text
  
      # The request should be hanging waiting on the `SignalActor`.
      first_ref = do_req.remote()
  
      def check_request_started(num_expected_requests: int) -> bool:
          with pytest.raises(TimeoutError):
              ray.get(first_ref, timeout=0.1)
          assert (
              ray.get(counter_actor.get.remote())
              == num_expected_requests
          )
          return True
  
      wait_for_condition(
          check_request_started, timeout=5, num_expected_requests=1
      )
  
      # Now issue 10 more requests and wait for significantly longer than the long poll
      # timeout. They should all be queued in the handle due to `max_concurrent_queries`
      # enforcement (verified via the counter).
      new_refs = [do_req.remote() for _ in range(10)]
      ready, _ = ray.wait(new_refs, timeout=1)
      assert len(ready) == 0
      assert ray.get(counter_actor.get.remote()) == 1
  
      # Unblock the first request. Now everything should get executed.
      ray.get(signal_actor.send.remote())
      assert ray.get(first_ref) == "hello"
      assert ray.get(new_refs) == ["hello"] * 10
      assert ray.get(counter_actor.get.remote()) == 11
  
      serve.shutdown()
  
imports: import os
  import subprocess
  import sys
  from contextlib import contextmanager
  from tempfile import NamedTemporaryFile
  import pytest
  import requests
  import ray
  import ray._private.state
  import ray.actor
  from ray import serve
  from ray._private.test_utils import SignalActor, wait_for_condition
  from ray.cluster_utils import AutoscalingCluster, Cluster
  from ray.exceptions import RayActorError
  from ray.serve._private.common import ProxyStatus
  from ray.serve._private.constants import SERVE_DEFAULT_APP_NAME
  from ray.serve._private.logging_utils import get_serve_logs_dir
  from ray.serve._private.utils import get_head_node_id
  from ray.serve.context import _get_global_client
  from ray.serve.schema import ServeInstanceDetails
  from ray.tests.conftest import call_ray_stop_only
label_desc: Use image node_id 4 for calling a global function with example usage: test_long_poll_timeout_with_max_concurrent_queries(ray_instance) and returns: str, requests, True, self