code: def test_in_flight_queued_requests_canceled(
      shutdown_only, monkeypatch
  ):
      """
      When there are large input size in-flight actor tasks
      tasks are queued inside a RPC layer (core_worker_client.h)
      In this case, we don't cancel a request from a client side
      but wait until it is sent to the server side and cancel it.
      See SendRequests() inside core_worker_client.h
      """
      # Currently the max bytes is
      # const int64_t kMaxBytesInFlight = 16 * 1024 * 1024.
      # See core_worker_client.h.
      input_arg = b"1" * 15 * 1024  # 15KB.
      # Tasks are queued when there are more than 1024 tasks.
      sig = SignalActor.remote()
  
      @ray.remote
      class Actor:
          def __init__(self, signal_actor):
              self.signal_actor = signal_actor
  
          def f(self, input_arg):
              ray.get(self.signal_actor.wait.remote())
              return True
  
      a = Actor.remote(sig)
      refs = [a.f.remote(input_arg) for _ in range(5000)]
  
      # Wait until the first task runs.
      wait_for_condition(
          lambda: len(list_tasks(filters=[("STATE", "=", "RUNNING")]))
          == 1
      )
  
      # Cancel all tasks.
      for ref in refs:
          ray.cancel(ref)
  
      # The first ref is in progress, so we pop it out
      first_ref = refs.pop(0)
      ray.get(sig.send.remote())
  
      # Make sure all tasks that are queued (including queued
      # due to in-flight bytes) are canceled.
      canceled = 0
      for ref in refs:
          try:
              ray.get(ref)
          except TaskCancelledError:
              canceled += 1
  
      # Verify at least half of tasks are canceled.
      # Currently, the guarantee is weak because we cannot
      # detect queued tasks due to inflight bytes limit.
      # TODO(sang): Move the in flight bytes logic into
      # actor submission queue instead of doing it inside
      # core worker client.
      assert canceled > 2500
  
      # first ref shouldn't have been canceled.
      assert ray.get(first_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 4 for calling a global function with example usage: test_in_flight_queued_requests_canceled(shutdown_only, monkeypatch) and returns: True