code: def test_recursive_cancel_actor_task(shutdown_only):
      ray.init()
  
      @ray.remote(num_cpus=0)
      class Semaphore:
          def wait(self):
              import time
  
              time.sleep(600)
  
      @ray.remote(num_cpus=0)
      class Actor2:
          def __init__(self, obj):
              (self.obj,) = obj
  
          def cancel(self):
              ray.cancel(self.obj)
  
      @ray.remote
      def task(sema):
          return ray.get(sema.wait.remote())
  
      sema = Semaphore.remote()
  
      t = task.remote(sema)
  
      def wait_until_wait_task_starts():
          wait_state = list_tasks(
              filters=[("func_or_class_name", "=", "Semaphore.wait")]
          )[0]
          return wait_state["state"] == "RUNNING"
  
      wait_for_condition(wait_until_wait_task_starts)
  
      # Make sure this will not crash ray.
      # https://github.com/ray-project/ray/issues/31398
      a2 = Actor2.remote((t,))
      a2.cancel.remote()
  
      with pytest.raises(RayTaskError, match="TaskCancelledError"):
          ray.get(t)
  
      wait_state = list_tasks(
          filters=[("func_or_class_name", "=", "Semaphore.wait")]
      )
      assert len(wait_state) == 1
      wait_state = wait_state[0]
      task_state = list_tasks(
          filters=[("func_or_class_name", "=", "task")]
      )
      assert len(task_state) == 1
      task_state = task_state[0]
  
      def verify():
          wait_state = list_tasks(
              filters=[("func_or_class_name", "=", "Semaphore.wait")]
          )
          assert len(wait_state) == 1
          wait_state = wait_state[0]
          task_state = list_tasks(
              filters=[("func_or_class_name", "=", "task")]
          )
          assert len(task_state) == 1
          task_state = task_state[0]
  
          assert task_state["state"] == "FINISHED"
          assert wait_state["state"] == "RUNNING"
  
          return True
  
      wait_for_condition(verify)
  
imports: import os
  import random
  import signal
  import sys
  import threading
  import _thread
  import time
  import pytest
  import ray
  from ray.exceptions import TaskCancelledError, RayTaskError, GetTimeoutError, WorkerCrashedError, ObjectLostError
  from ray._private.utils import DeferSigint
  from ray._private.test_utils import SignalActor, wait_for_condition
  from ray.util.state import list_tasks
label_desc: Use image node_id 15 for calling a global function with example usage: test_recursive_cancel_actor_task(shutdown_only) and returns: ray, unknown, True