code: def test_async_actor_client_side_cancel(ray_start_cluster):
      """
      Test a task is cancelled while it is queued on a client side.
      It should raise ray.exceptions.TaskCancelledError.
      """
      cluster = ray_start_cluster
      cluster.add_node(num_cpus=0)
      ray.init(address=cluster.address)
  
      @ray.remote(num_cpus=1)
      class Actor:
          def __init__(self):
              self.f_called = False
  
          async def g(self, ref):
              await asyncio.sleep(30)
  
          async def f(self):
              self.f_called = True
              await asyncio.sleep(5)
  
          def is_f_called(self):
              return self.f_called
  
      @ray.remote
      def f():
          time.sleep(100)
  
      # Test the case where a task is queued on a client side.
      # Tasks are not sent until actor is created.
      a = Actor.remote()
      ref = a.f.remote()
      ray.cancel(ref)
      with pytest.raises(TaskCancelledError):
          ray.get(ref)
  
      cluster.add_node(num_cpus=1)
      assert not ray.get(a.is_f_called.remote())
  
      # Test the case where it is canceled before dependencies
      # are resolved.
      a = Actor.remote()
      ref_dep_not_resolved = a.g.remote(f.remote())
      ray.cancel(ref_dep_not_resolved)
      with pytest.raises(TaskCancelledError):
          ray.get(ref_dep_not_resolved)
  
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 3 for calling a global function with example usage: test_async_actor_client_side_cancel(ray_start_cluster) and returns: self