code: def test_cancel_recursive_chain(shutdown_only, recursive):
      @ray.remote
      class RecursiveActor:
          def __init__(self, child=None):
              self.child = child
              self.chlid_ref = None
  
          async def run(self, sig):
              if self.child is None:
                  await sig.wait.remote()
                  return True
  
              ref = self.child.run.remote(sig)
              self.child_ref = ref
              return await ref
  
          def get_child_ref(self):
              return self.child_ref
  
      sig = SignalActor.remote()
      r1 = RecursiveActor.remote()
      r2 = RecursiveActor.remote(r1)
      r3 = RecursiveActor.remote(r2)
      r4 = RecursiveActor.remote(r3)
  
      ref = r4.run.remote(sig)
      ray.get(r4.__ray_ready__.remote())
      wait_for_condition(
          lambda: len(
              list_tasks(filters=[("name", "=", "RecursiveActor.run")])
          )
          == 4
      )
      ray.cancel(ref, recursive=recursive)
      ray.get(sig.send.remote())
  
      if recursive:
          with pytest.raises(ray.exceptions.TaskCancelledError):
              ray.get(ref)
          with pytest.raises(ray.exceptions.TaskCancelledError):
              ray.get(ray.get(r4.get_child_ref.remote()))
          with pytest.raises(ray.exceptions.TaskCancelledError):
              ray.get(ray.get(r3.get_child_ref.remote()))
          with pytest.raises(ray.exceptions.TaskCancelledError):
              ray.get(ray.get(r2.get_child_ref.remote()))
      else:
          assert ray.get(ray.get(r2.get_child_ref.remote()))
          assert ray.get(ray.get(r3.get_child_ref.remote()))
          assert ray.get(ray.get(r4.get_child_ref.remote()))
  
          with pytest.raises(ray.exceptions.TaskCancelledError):
              ray.get(ref)
  
imports: import asyncio
  import os
  import sys
  import time
  from collections import defaultdict
  import pytest
  import ray
  from ray._private.test_utils import SignalActor, wait_for_condition
  from ray.exceptions import TaskCancelledError
  from ray.util.state import list_tasks
label_desc: Use image node_id 11 for calling a global function with example usage: test_cancel_recursive_chain(shutdown_only, recursive) and returns: unknown, self, True