code: def test_worker_startup_count(ray_start_cluster):
      """Test that no extra workers started while no available cpu resources
      in cluster."""
  
      cluster = ray_start_cluster
      # Cluster total cpu resources is 4.
      cluster.add_node(
          num_cpus=4,
          _system_config={
              "debug_dump_period_milliseconds": 100,
          },
      )
      ray.init(address=cluster.address)
  
      # A slow function never returns. It will hold cpu resources all the way.
      @ray.remote
      def slow_function():
          while True:
              time.sleep(1000)
  
      # Flood a large scale lease worker requests.
      for i in range(10000):
          slow_function.options(num_cpus=0.25).remote()
  
      # Check "debug_state.txt" to ensure no extra workers were started.
      session_dir = ray._private.worker.global_worker.node.address_info[
          "session_dir"
      ]
      session_path = Path(session_dir)
      debug_state_path = session_path / "logs" / "debug_state.txt"
  
      def get_num_workers():
          with open(debug_state_path) as f:
              for line in f.readlines():
                  num_workers_prefix = "- num PYTHON workers: "
                  if num_workers_prefix in line:
                      num_workers = int(line[len(num_workers_prefix) :])
                      return num_workers
          return None
  
      # Wait for "debug_state.txt" to be updated to reflect the started worker.
      timeout_limit = 15
      start = time.time()
      wait_for_condition(
          lambda: get_num_workers() == 16, timeout=timeout_limit
      )
      time_waited = time.time() - start
      print(f"Waited {time_waited} for debug_state.txt to be updated")
  
      # Check that no more workers started for a while.
      for i in range(100):
          # Sometimes the debug state file can be empty. Retry if needed.
          for _ in range(3):
              num = get_num_workers()
              if num is None:
                  print("Retrying parse debug_state.txt")
                  time.sleep(0.05)
              else:
                  break
          assert num == 16
          time.sleep(0.1)
  
imports: import logging
  import subprocess
  import sys
  import time
  from pathlib import Path
  import os
  import pytest
  from unittest import mock
  import ray
  import ray.cluster_utils
  from ray._private.test_utils import wait_for_condition
  from ray.autoscaler._private.constants import RAY_PROCESSES
  import psutil
label_desc: Use image node_id 2 for calling a global function with example usage: test_worker_startup_count(ray_start_cluster) and returns: None, num_workers