code: def test_reconfigure_multiple_replicas(serve_instance, use_handle):
      # Tests that updating the user_config with multiple replicas performs a
      # rolling update.
      client = serve_instance
  
      name = "test"
  
      @ray.remote(num_cpus=0)
      def call():
          if use_handle:
              handle = serve.get_deployment_handle(name, "app")
              ret = handle.handler.remote().result()
          else:
              ret = requests.get(f"http://localhost:8000/{name}").text
  
          return ret.split("|")[0], ret.split("|")[1]
  
      signal_name = f"signal-{get_random_string()}"
      signal = SignalActor.options(name=signal_name).remote()
  
      @serve.deployment(name=name, version="1", num_replicas=2)
      class V1:
          def __init__(self):
              self.config = None
  
          async def reconfigure(self, config):
              # Don't block when the replica is first created.
              if self.config is not None:
                  signal = ray.get_actor(signal_name)
                  ray.get(signal.wait.remote())
              self.config = config
  
          async def handler(self):
              val = self.config["test"]
              return f"{val}|{os.getpid()}"
  
          async def __call__(self, request):
              return await self.handler()
  
      def make_nonblocking_calls(expected, expect_blocking=False):
          # Returns dict[val, set(pid)].
          blocking = []
          responses = defaultdict(set)
          start = time.time()
          while time.time() - start < 30:
              refs = [call.remote() for _ in range(10)]
              ready, not_ready = ray.wait(refs, timeout=5)
              for ref in ready:
                  val, pid = ray.get(ref)
                  responses[val].add(pid)
              for ref in not_ready:
                  blocking.extend(not_ready)
  
              if all(
                  len(responses[val]) == num
                  for val, num in expected.items()
              ) and (expect_blocking is False or len(blocking) > 0):
                  break
          else:
              assert False, f"Timed out, responses: {responses}."
  
          return responses, blocking
  
      serve.run(
          V1.options(user_config={"test": "1"}).bind(), name="app"
      )
      responses1, _ = make_nonblocking_calls({"1": 2})
      pids1 = responses1["1"]
  
      # Reconfigure should block one replica until the signal is sent. Check that
      # some requests are now blocking.
      serve.run(
          V1.options(user_config={"test": "2"}).bind(),
          name="app",
          _blocking=False,
      )
      responses2, blocking2 = make_nonblocking_calls(
          {"1": 1}, expect_blocking=True
      )
      assert list(responses2["1"])[0] in pids1
  
      # Signal reconfigure to finish. Now the goal should complete and both
      # replicas should have the updated config.
      ray.get(signal.send.remote())
      client._wait_for_application_running("app")
      make_nonblocking_calls({"2": 2})
  
imports: import os
  import sys
  import time
  from collections import defaultdict
  from typing import Callable
  import pytest
  import requests
  import ray
  from ray import serve
  from ray._private.pydantic_compat import ValidationError
  from ray._private.test_utils import SignalActor
  from ray.serve._private.utils import get_random_string
  from ray.serve.exceptions import RayServeException
label_desc: Use image node_id 6 for calling a global function with example usage: test_reconfigure_multiple_replicas(serve_instance, use_handle) and returns: ret,  ret, responses,  blocking, val, str, os, unknown