code: def test_many_queued_tasks():
      sema = Semaphore.remote(0)
  
      @ray.remote(num_cpus=1)
      def block():
          ray.get(sema.acquire.remote())
  
      @ray.remote(num_cpus=1)
      def f():
          pass
  
      num_cpus = int(ray.cluster_resources()["CPU"])
      blocked_tasks = []
      for _ in range(num_cpus):
          blocked_tasks.append(block.remote())
  
      print("Submitting many tasks")
      pending_tasks = []
      for _ in trange(MAX_QUEUED_TASKS):
          pending_tasks.append(f.remote())
  
      # Make sure all the tasks can actually run.
      for _ in range(num_cpus):
          sema.release.remote()
  
      print("Unblocking tasks")
      for ref in tqdm(pending_tasks):
          assert ray.get(ref) is None
  
file_path: ray/release/benchmarks/single_node/test_single_node.py
imports: import numpy
  import time
  import ray
  import ray.autoscaler.sdk
  from ray._private.test_utils import Semaphore
  import json
  import os
  from time import perf_counter
  from tqdm import trange, tqdm
label_desc: Use image node_id 5 for calling a global function with example usage: test_many_queued_tasks() without return types