code: def test_redeploy_multiple_replicas(serve_instance, use_handle):
      # Tests that redeploying a deployment with multiple replicas performs
      # a rolling update.
      client = serve_instance
  
      name = "test"
  
      @ray.remote(num_cpus=0)
      def call(block=False):
          if use_handle:
              handle = serve.get_deployment_handle(name, "app")
              ret = handle.handler.remote(block).result()
          else:
              ret = requests.get(
                  f"http://localhost:8000/{name}",
                  params={"block": block},
              ).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:
          async def handler(self, block: bool):
              if block:
                  signal = ray.get_actor(signal_name)
                  await signal.wait.remote()
  
              return f"1|{os.getpid()}"
  
          async def __call__(self, request):
              return await self.handler(
                  request.query_params["block"] == "True"
              )
  
      class V2:
          async def handler(self, *args):
              return f"2|{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(block=False) 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.bind(), name="app")
      responses1, _ = make_nonblocking_calls({"1": 2})
      pids1 = responses1["1"]
  
      # ref2 will block a single replica until the signal is sent. Check that
      # some requests are now blocking.
      ref2 = call.remote(block=True)
      responses2, blocking2 = make_nonblocking_calls(
          {"1": 1}, expect_blocking=True
      )
      assert list(responses2["1"])[0] in pids1
  
      # Redeploy new version. Since there is one replica blocking, only one new
      # replica should be started up.
      V2 = V1.options(func_or_class=V2, version="2")
      serve.run(V2.bind(), _blocking=False, name="app")
      with pytest.raises(TimeoutError):
          client._wait_for_application_running("app", timeout_s=0.1)
      responses3, blocking3 = make_nonblocking_calls(
          {"1": 1}, expect_blocking=True
      )
  
      # Signal the original call to exit.
      ray.get(signal.send.remote())
      val, pid = ray.get(ref2)
      assert val == "1"
      assert pid in responses1["1"]
  
      # Now the goal and requests to the new version should complete.
      # We should have two running replicas of the new version.
      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 5 for calling a global function with example usage: test_redeploy_multiple_replicas(serve_instance, use_handle) and returns: ret,  ret, responses,  blocking, str, os, unknown, str, os, unknown