code: def test_queued_queries_disconnected(serve_start_shutdown):
      """Check that disconnected queued queries are tracked correctly."""
  
      signal = SignalActor.remote()
  
      @serve.deployment(
          max_concurrent_queries=1,
      )
      async def hang_on_first_request():
          await signal.wait.remote()
  
      serve.run(hang_on_first_request.bind())
  
      print("Deployed hang_on_first_request deployment.")
  
      wait_for_condition(
          check_metric_float_eq,
          timeout=15,
          metric="ray_serve_num_scheduling_tasks",
          expected=-1,  # -1 means not expected to be present yet.
      )
      print("ray_serve_num_scheduling_tasks updated successfully.")
      wait_for_condition(
          check_metric_float_eq,
          timeout=15,
          metric="serve_num_scheduling_tasks_in_backoff",
          expected=-1,  # -1 means not expected to be present yet.
      )
      print(
          "serve_num_scheduling_tasks_in_backoff updated successfully."
      )
  
      @ray.remote(num_cpus=0)
      def do_request():
          r = requests.get("http://localhost:8000/")
          r.raise_for_status()
          return r
  
      # Make a request to block the deployment from accepting other requests.
      request_refs = [do_request.remote()]
      wait_for_condition(
          lambda: ray.get(signal.cur_num_waiters.remote()) == 1,
          timeout=10,
      )
  
      print("First request is executing.")
      wait_for_condition(
          check_metric_float_eq,
          timeout=15,
          metric="ray_serve_num_ongoing_http_requests",
          expected=1,
      )
      print("ray_serve_num_ongoing_http_requests updated successfully.")
  
      num_queued_requests = 3
      request_refs.extend(
          [do_request.remote() for _ in range(num_queued_requests)]
      )
      print(f"{num_queued_requests} more requests now queued.")
  
      # First request should be processing. All others should be queued.
      wait_for_condition(
          check_metric_float_eq,
          timeout=15,
          metric="ray_serve_deployment_queued_queries",
          expected=num_queued_requests,
      )
      print("ray_serve_deployment_queued_queries updated successfully.")
      wait_for_condition(
          check_metric_float_eq,
          timeout=15,
          metric="ray_serve_num_ongoing_http_requests",
          expected=num_queued_requests + 1,
      )
      print("ray_serve_num_ongoing_http_requests updated successfully.")
  
      # There should be 2 scheduling tasks (which is the max, since
      # 2 = 2 * 1 replica) that are attempting to schedule the hanging requests.
      wait_for_condition(
          check_metric_float_eq,
          timeout=15,
          metric="ray_serve_num_scheduling_tasks",
          expected=2,
      )
      print("ray_serve_num_scheduling_tasks updated successfully.")
      wait_for_condition(
          check_metric_float_eq,
          timeout=15,
          metric="serve_num_scheduling_tasks_in_backoff",
          expected=2,
      )
      print(
          "serve_num_scheduling_tasks_in_backoff updated successfully."
      )
  
      # Disconnect all requests by cancelling the Ray tasks.
      [ray.cancel(ref, force=True) for ref in request_refs]
      print("Cancelled all HTTP requests.")
  
      wait_for_condition(
          check_metric_float_eq,
          timeout=15,
          metric="ray_serve_deployment_queued_queries",
          expected=0,
      )
      print("ray_serve_deployment_queued_queries updated successfully.")
  
      # Task should get cancelled.
      wait_for_condition(
          check_metric_float_eq,
          timeout=15,
          metric="ray_serve_num_ongoing_http_requests",
          expected=0,
      )
      print("ray_serve_num_ongoing_http_requests updated successfully.")
  
      wait_for_condition(
          check_metric_float_eq,
          timeout=15,
          metric="ray_serve_num_scheduling_tasks",
          expected=0,
      )
      print("ray_serve_num_scheduling_tasks updated successfully.")
      wait_for_condition(
          check_metric_float_eq,
          timeout=15,
          metric="serve_num_scheduling_tasks_in_backoff",
          expected=0,
      )
      print(
          "serve_num_scheduling_tasks_in_backoff updated successfully."
      )
  
      # Unblock hanging request.
      ray.get(signal.send.remote())
  
imports: import os
  import sys
  from typing import DefaultDict, Dict, List, Optional
  import grpc
  import pytest
  import requests
  from fastapi import FastAPI
  import ray
  import ray.util.state
  from ray import serve
  from ray._private.test_utils import SignalActor, wait_for_condition
  from ray.serve._private.constants import DEFAULT_LATENCY_BUCKET_MS
  from ray.serve._private.long_poll import LongPollHost, UpdatedObject
  from ray.serve._private.test_utils import ping_fruit_stand, ping_grpc_call_method, ping_grpc_list_applications
  from ray.serve._private.utils import block_until_http_ready
  from ray.serve.config import gRPCOptions
  from ray.serve.handle import DeploymentHandle
  from ray.serve.metrics import Counter, Gauge, Histogram
  from ray.serve.tests.test_config_files.grpc_deployment import g, g2
label_desc: Use image node_id 10 for calling a global function with example usage: test_queued_queries_disconnected(serve_start_shutdown) and returns: r