code: def test_async_actor_cancel(shutdown_only):
      """
      Test async actor task is canceled and
      asyncio.CancelledError is raised within a task.
      """
      ray.init(num_cpus=1)
  
      @ray.remote
      class VerifyActor:
          def __init__(self):
              self.called = False
              self.running = False
  
          def called(self):
              self.called = True
  
          def set_running(self):
              self.running = True
  
          def is_called(self):
              return self.called
  
          def is_running(self):
              return self.running
  
          def reset(self):
              self.called = False
              self.running = False
  
      @ray.remote
      class Actor:
          async def f(self, verify_actor):
              try:
                  ray.get(verify_actor.set_running.remote())
                  await asyncio.sleep(10)
              except asyncio.CancelledError:
                  # It is False until this except block is finished.
                  print(asyncio.current_task().cancelled())
                  assert not asyncio.current_task().cancelled()
                  ray.get(verify_actor.called.remote())
                  raise
              except Exception:
                  return True
              return True
  
      v = VerifyActor.remote()
      a = Actor.remote()
  
      for i in range(50):
          ref = a.f.remote(v)
          wait_for_condition(lambda: ray.get(v.is_running.remote()))
          ray.cancel(ref)
  
          with pytest.raises(
              ray.exceptions.TaskCancelledError, match="was cancelled"
          ):
              ray.get(ref)
  
          # Verify asyncio.CancelledError is raised from the actor task.
          assert ray.get(v.is_running.remote())
          assert ray.get(v.is_called.remote())
          ray.get(v.reset.remote())
  
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 2 for calling a global function with example usage: test_async_actor_cancel(shutdown_only) and returns: self, self, True, True