code: def test_handle_prefers_replicas_on_same_node(ray_cluster):
      """Verify that handle calls prefer replicas on the same node when possible.
  
      If all replicas on the same node are occupied (at `max_concurrent_queries` limit),
      requests should spill to other nodes.
      """
  
      cluster = ray_cluster
      cluster.add_node(num_cpus=1)
      cluster.add_node(num_cpus=1)
  
      signal = SignalActor.remote()
  
      @serve.deployment(num_replicas=2, max_concurrent_queries=1)
      def inner(block_on_signal):
          if block_on_signal:
              ray.get(signal.wait.remote())
  
          return ray.get_runtime_context().get_node_id()
  
      @serve.deployment(
          num_replicas=1, ray_actor_options={"num_cpus": 0}
      )
      class Outer:
          def __init__(self, inner_handle: RayServeHandle):
              self._h = inner_handle.options(_prefer_local_routing=True)
  
          def get_node_id(self) -> str:
              return ray.get_runtime_context().get_node_id()
  
          async def call_inner(
              self, block_on_signal: bool = False
          ) -> str:
              return await self._h.remote(block_on_signal)
  
      # The inner deployment's two replicas will be spread across the two nodes and
      # the outer deployment's single replica will be placed on one of them.
      h = serve.run(Outer.bind(inner.bind()))
  
      # When sending requests sequentially, all requests to the inner deployment should
      # go to the replica on the same node as the outer deployment replica.
      outer_node_id = h.get_node_id.remote().result()
      for _ in range(10):
          assert h.call_inner.remote().result() == outer_node_id
  
      # Make a blocking request to the inner deployment replica on the same node.
      blocked_response = h.call_inner.remote(block_on_signal=True)
      with pytest.raises(TimeoutError):
          blocked_response.result(timeout_s=1)
  
      # Because there's a blocking request and `max_concurrent_queries` is set to 1, all
      # requests should now spill to the other node.
      for _ in range(10):
          assert h.call_inner.remote().result() != outer_node_id
  
      ray.get(signal.send.remote())
      assert blocked_response.result() == outer_node_id
  
imports: import os
  import sys
  import time
  from collections import defaultdict
  import pytest
  import requests
  import ray
  from ray import serve
  from ray._private.test_utils import SignalActor, wait_for_condition
  from ray.cluster_utils import Cluster
  from ray.exceptions import RayActorError
  from ray.serve._private.common import DeploymentID, ReplicaState
  from ray.serve._private.constants import SERVE_NAMESPACE
  from ray.serve._private.deployment_state import ReplicaStartupStatus
  from ray.serve._private.utils import get_head_node_id
  from ray.serve.context import _get_global_client
  from ray.serve.handle import RayServeHandle
  from ray.serve.schema import ServeDeploySchema
label_desc: Use image node_id 9 for calling a global function with example usage: test_handle_prefers_replicas_on_same_node(ray_cluster) and returns: ray, ray, unknown