code: def test_fast(shutdown_only, use_force):
      ray.init(num_cpus=2)
  
      @ray.remote
      def fast(y):
          return y
  
      signaler = SignalActor.remote()
      ids = list()
      for _ in range(100):
          x = fast.remote("a")
          # NOTE If a non-force Cancellation is attempted in the time
          # between a worker receiving a task and the worker executing
          # that task (specifically the python execution), Cancellation
          # can fail.
  
          time.sleep(0.1)
          ray.cancel(x, force=use_force)
          ids.append(x)
  
      @ray.remote
      def wait_for(y):
          return y
  
      sig = signaler.wait.remote()
      for _ in range(5000):
          x = wait_for.remote(sig)
          ids.append(x)
  
      for idx in range(100, 5100):
          if random.random() > 0.95:
              ray.cancel(ids[idx], force=use_force)
      signaler.send.remote()
      for i, obj_ref in enumerate(ids):
          try:
              ray.get(obj_ref, timeout=120)
          except Exception as e:
              assert isinstance(
                  e, valid_exceptions(use_force)
              ), f"Failure on iteration: {i}"
  
imports: import os
  import random
  import signal
  import sys
  import threading
  import _thread
  import time
  import pytest
  import ray
  from ray.exceptions import TaskCancelledError, RayTaskError, GetTimeoutError, WorkerCrashedError, ObjectLostError
  from ray._private.utils import DeferSigint
  from ray._private.test_utils import SignalActor, wait_for_condition
  from ray.util.state import list_tasks
label_desc: Use image node_id 12 for calling a global function with example usage: test_fast(shutdown_only, use_force) and returns: y, y