code: def test_actor_failure_async_4(ray_start_regular, tmp_path):
      from filelock import FileLock
  
      l_file = tmp_path / "lock"
  
      l_lock = FileLock(l_file)
      l_lock.acquire()
  
      @ray.remote
      def f():
          with FileLock(l_file):
              os.kill(os.getpid(), SIGKILL)
  
      @ray.remote(max_restarts=1)
      class A:
          def pid(self, x):
              return os.getpid()
  
      a = A.remote()
  
      def new_task(_):
          print("new_task")
          # make sure there is no deadlock
          a.pid.remote(None)
  
      t = a.pid.remote(f.remote())
      # Make sure there is no deadlock when executing
      # the callback
      t._on_completed(new_task)
  
      ray.kill(a)
  
      # This will make the dependence failed
      l_lock.release()
  
      with pytest.raises(Exception):
          ray.get(t)
  
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 7 for calling a global function with example usage: test_actor_failure_async_4(ray_start_regular, tmp_path) and returns: os