code: def test_caller_actor_restart(ray_start_regular):
      """Test tasks from a restarted actor can be correctly processed
      by the receiving actor."""
  
      @ray.remote(max_restarts=1, max_task_retries=-1)
      class RestartableActor:
          """An actor that will be restarted at most once."""
  
          def __init__(self, actor):
              self.actor = actor
  
          def increase(self):
              return ray.get(self.actor.increase.remote())
  
          def get_pid(self):
              return os.getpid()
  
      @ray.remote(max_restarts=1)
      class Actor:
          """An actor that will be restarted at most once."""
  
          def __init__(self):
              self.value = 0
  
          def increase(self):
              self.value += 1
              return self.value
  
      remote_actor = Actor.remote()
      actor = RestartableActor.remote(remote_actor)
      # Call increase 3 times
      for _ in range(3):
          ray.get(actor.increase.remote())
  
      # kill the actor.
      # TODO(zhijunfu): use ray.kill instead.
      kill_actor(actor)
  
      # Check that we can still call the actor.
      assert ray.get(actor.increase.remote()) == 4
  
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 7 for calling a global function with example usage: test_caller_actor_restart(ray_start_regular) and returns: ray, os, self