code: def test_actor_restart(ray_init_with_task_retry_delay):
      """Test actor restart when actor process is killed."""
  
      @ray.remote(max_restarts=1)
      class RestartableActor:
          """An actor that will be restarted at most once."""
  
          def __init__(self):
              self.value = 0
  
          def increase(self, exit=False):
              if exit:
                  os._exit(-1)
              self.value += 1
              return self.value
  
          def get_pid(self):
              return os.getpid()
  
      actor = RestartableActor.remote()
      # Submit some tasks and kill on a task midway through.
      results = [
          actor.increase.remote(exit=(i == 100)) for i in range(200)
      ]
      # Make sure that all tasks were executed in order before the actor's death.
      i = 1
      while results:
          res = results[0]
          try:
              r = ray.get(res)
              if r != i:
                  # Actor restarted at this task without any failed tasks in
                  # between.
                  break
              results.pop(0)
              i += 1
          except ray.exceptions.RayActorError:
              break
      # Skip any tasks that errored.
      while results:
          try:
              ray.get(results[0])
          except ray.exceptions.RayActorError:
              results.pop(0)
          else:
              break
      # Check all tasks that executed after the restart.
      if results:
          # The actor executed some tasks after the restart.
          i = 1
          while results:
              r = ray.get(results.pop(0))
              assert r == i
              i += 1
  
          # Check that we can still call the actor.
          result = actor.increase.remote()
          assert ray.get(result) == r + 1
      else:
          # Wait for the actor to restart.
          def ping():
              try:
                  ray.get(actor.increase.remote())
                  return True
              except ray.exceptions.RayActorError:
                  return False
  
          wait_for_condition(ping)
  
      # The actor has restarted. Kill actor process one more time.
      actor.increase.remote(exit=True)
      # The actor has exceeded max restarts. All tasks should fail.
      for _ in range(100):
          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 3 for calling a global function with example usage: test_actor_restart(ray_init_with_task_retry_delay) and returns: self, os, True, False