code: def test_actor_failure_async_2(ray_start_regular, tmp_path):
      p = tmp_path / "a_pid"
  
      @ray.remote(max_restarts=1)
      class A:
          def __init__(self):
              pid = os.getpid()
              # The second time start, it'll block,
              # so that we'll know the actor is restarting.
              if p.exists():
                  p.write_text(str(pid))
                  time.sleep(100000)
              else:
                  p.write_text(str(pid))
  
          def pid(self):
              return os.getpid()
  
      a = A.remote()
  
      pid = ray.get(a.pid.remote())
  
      os.kill(int(pid), SIGKILL)
  
      # kill will be in another thred.
      def kill():
          # sleep for 2s for the code to be setup
          time.sleep(2)
          new_pid = int(p.read_text())
          while new_pid == pid:
              new_pid = int(p.read_text())
              time.sleep(1)
          os.kill(new_pid, SIGKILL)
  
      t = threading.Thread(target=kill)
      t.start()
  
      try:
          o = a.pid.remote()
  
          def new_task(_):
              print("new_task")
              # make sure there is no deadlock
              a.pid.remote()
  
          o._on_completed(new_task)
          # When ray.get(o) failed,
          # new_task will be executed
          ray.get(o)
      except Exception:
          pass
      t.join()
  
imports: import os
  import sys
  import signal
  import threading
  import json
  from pathlib import Path
  import ray
  import numpy
  import pytest
  import psutil
  import time
  from ray._private.test_utils import SignalActor, wait_for_pid_to_exit, wait_for_condition, run_string_as_driver_nonblocking
label_desc: Use image node_id 5 for calling a global function with example usage: test_actor_failure_async_2(ray_start_regular, tmp_path) and returns: os