code: def test_actor_restart_with_retry(ray_init_with_task_retry_delay):
      """Test actor restart when actor process is killed."""
  
      @ray.remote(max_restarts=1, max_task_retries=-1)
      class RestartableActor:
          """An actor that will be restarted at most once."""
  
          def __init__(self):
              self.value = 0
  
          def increase(self, delay=0):
              time.sleep(delay)
              self.value += 1
              return self.value
  
          def get_pid(self):
              return os.getpid()
  
      actor = RestartableActor.remote()
      pid = ray.get(actor.get_pid.remote())
      results = [actor.increase.remote() for _ in range(100)]
      # Kill actor process, while the above task is still being executed.
      os.kill(pid, SIGKILL)
      wait_for_pid_to_exit(pid)
      # Check that none of the tasks failed and the actor is restarted.
      seq = list(range(1, 101))
      results = ray.get(results)
      failed_task_index = None
      # Make sure that all tasks were executed in order before and after the
      # actor's death.
      for i, res in enumerate(results):
          if res != seq[0]:
              if failed_task_index is None:
                  failed_task_index = i
              assert res + failed_task_index == seq[0]
          seq.pop(0)
      # Check that we can still call the actor.
      result = actor.increase.remote()
      assert ray.get(result) == results[-1] + 1
  
      # kill actor process one more time.
      results = [actor.increase.remote() for _ in range(100)]
      pid = ray.get(actor.get_pid.remote())
      os.kill(pid, SIGKILL)
      wait_for_pid_to_exit(pid)
      # The actor has exceeded max restarts, and this task should fail.
      with pytest.raises(ray.exceptions.RayActorError):
          ray.get(actor.increase.remote())
  
      # Create another actor.
      actor = RestartableActor.remote()
      # Intentionlly exit the actor
      actor.__ray_terminate__.remote()
      # Check that the actor won't be restarted.
      with pytest.raises(ray.exceptions.RayActorError):
          ray.get(actor.increase.remote())
  
imports: import atexit
  import asyncio
  import collections
  import numpy
  import os
  import pytest
  import signal
  import sys
  import time
  import ray
  from ray.actor import exit_actor
  import ray.cluster_utils
  from ray._private.test_utils import wait_for_condition, wait_for_pid_to_exit, generate_system_config_map, SignalActor
label_desc: Use image node_id 4 for calling a global function with example usage: test_actor_restart_with_retry(ray_init_with_task_retry_delay) and returns: self, os