code: def test_recover_rolling_update_from_replica_actor_names(
      serve_instance,
  ):
      """Test controller is able to recover starting -> updating -> running
      replicas from actor names, with right replica versions during rolling
      update.
      """
      client = serve_instance
  
      name = "test"
  
      @ray.remote(num_cpus=0)
      def call(block=False):
          handle = serve.get_deployment_handle(name, "app")
          ret = handle.handler.remote(block).result()
  
          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, num_returns=1
      ):
          # Returns dict[val, set(pid)].
          blocking = []
          responses = defaultdict(set)
          start = time.time()
          timeout_value = 60 if sys.platform == "win32" else 30
          while time.time() - start < timeout_value:
              refs = [call.remote(block=False) for _ in range(10)]
              ready, not_ready = ray.wait(
                  refs, timeout=5, num_returns=num_returns
              )
              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}, num_returns=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
  
      ray.kill(
          serve.context._global_client._controller, no_restart=False
      )
  
      # 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
      )
  
      ray.kill(
          serve.context._global_client._controller, no_restart=False
      )
  
      # 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}, num_returns=2)
  
imports: import asyncio
  import logging
  import os
  import re
  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.exceptions import RayTaskError
  from ray.serve._private.common import DeploymentID, ReplicaState
  from ray.serve._private.constants import SERVE_CONTROLLER_NAME, SERVE_DEFAULT_APP_NAME, SERVE_NAMESPACE, SERVE_PROXY_NAME
  from ray.serve._private.utils import get_random_string
  from ray.serve.schema import LoggingConfig
  from ray.serve.tests.test_failure import request_with_retries
  from ray.util.state import list_actors
label_desc: Use image node_id 2 for calling a global function with example usage: test_recover_rolling_update_from_replica_actor_names(serve_instance) and returns: ret,  ret, responses,  blocking, str, os, unknown, str, os, unknown