code: def test_max_running_tasks(num_tasks):
      cpus_per_task = 0.25
  
      @ray.remote(num_cpus=cpus_per_task)
      def task():
          time.sleep(sleep_time)
  
      def time_up(start_time):
          return time.time() - start_time >= sleep_time
  
      refs = [
          task.remote()
          for _ in tqdm.trange(num_tasks, desc="Launching tasks")
      ]
  
      max_cpus = ray.cluster_resources()["CPU"]
      min_cpus_available = max_cpus
      start_time = time.time()
      for _ in tqdm.trange(int(sleep_time / 0.1), desc="Waiting"):
          try:
              cur_cpus = ray.available_resources().get("CPU", 0)
              min_cpus_available = min(min_cpus_available, cur_cpus)
          except Exception:
              # There are race conditions `.get` can fail if a new heartbeat
              # comes at the same time.
              pass
          if time_up(start_time):
              print(f"Time up for sleeping {sleep_time} seconds")
              break
          time.sleep(0.1)
  
      # There are some relevant magic numbers in this check. 10k tasks each
      # require 1/4 cpus. Therefore, ideally 2.5k cpus will be used.
      used_cpus = max_cpus - min_cpus_available
      err_str = f"Only {used_cpus}/{max_cpus} cpus used."
      # 1500 tasks. Note that it is a pretty low threshold, and the
      # performance should be tracked via perf dashboard.
      threshold = num_tasks * cpus_per_task * 0.60
      print(f"{used_cpus}/{max_cpus} used.")
      assert used_cpus > threshold, err_str
  
      for _ in tqdm.trange(
          num_tasks, desc="Ensuring all tasks have finished"
      ):
          done, refs = ray.wait(refs)
          assert ray.get(done[0]) is None
  
      return used_cpus
  
file_path: ray/release/benchmarks/distributed/test_many_tasks.py
imports: import click
  import ray
  import ray._private.test_utils
  import time
  import tqdm
  from ray.util.state import summarize_tasks
  from dashboard_test import DashboardTestAtScale
  from ray._private.state_api_test_utils import StateAPICallSpec, periodic_invoke_state_apis_with_actor, summarize_worker_startup_time
label_desc: Use image node_id 1 for calling a global function with example usage: test_max_running_tasks(num_tasks) and returns: used_cpus, unknown