code: def test_async_actor_server_side_cancel(shutdown_only):
      """
      Test Cancelation when a task is queued on a server side.
      """
  
      @ray.remote
      class Actor:
          async def f(self):
              await asyncio.sleep(5)
  
          async def g(self):
              await asyncio.sleep(0)
  
      a = Actor.options(max_concurrency=1).remote()
      ray.get(a.__ray_ready__.remote())
      ref = a.f.remote()  # noqa
      # Queued on a server side.
      # Task should not be executed at all.
      refs = [a.g.remote() for _ in range(100)]
      wait_for_condition(
          lambda: len(
              list_tasks(
                  filters=[
                      ("name", "=", "Actor.g"),
                      ("STATE", "=", "SUBMITTED_TO_WORKER"),
                  ]
              )
          )
          == 100
      )
  
      for ref in refs:
          ray.cancel(ref)
      tasks = list_tasks(filters=[("name", "=", "Actor.g")])
  
      for ref in refs:
          with pytest.raises(
              TaskCancelledError, match=ref.task_id().hex()
          ):
              ray.get(ref)
  
      # Verify the task is submitted to the worker and never executed
      # assert task.state == "SUBMITTED_TO_WORKER"
      for task in tasks:
          assert task.state == "SUBMITTED_TO_WORKER"
  
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 5 for calling a global function with example usage: test_async_actor_server_side_cancel(shutdown_only) without return types