code: def test_warning_for_too_many_nested_tasks(shutdown_only):
      # Check that if we run a workload which requires too many workers to be
      # started that we will receive a warning.
      num_cpus = 2
      ray.init(num_cpus=num_cpus)
      p = init_error_pubsub()
  
      remote_wait = Semaphore.remote(value=0)
      nested_wait = Semaphore.remote(value=0)
  
      ray.get(
          [
              remote_wait.locked.remote(),
              nested_wait.locked.remote(),
          ]
      )
  
      @ray.remote(num_cpus=0.25)
      def f():
          time.sleep(1000)
          return 1
  
      @ray.remote(num_cpus=0.25)
      def h(nested_waits):
          nested_wait.release.remote()
          ray.get(nested_waits)
          ray.get(f.remote())
  
      @ray.remote(num_cpus=0.25)
      def g(remote_waits, nested_waits):
          # Sleep so that the f tasks all get submitted to the scheduler after
          # the g tasks.
          remote_wait.release.remote()
          # wait until every lock is released.
          ray.get(remote_waits)
          ray.get(h.remote(nested_waits))
  
      num_root_tasks = num_cpus * 4
      # Lock remote task until everything is scheduled.
      remote_waits = []
      nested_waits = []
      for _ in range(num_root_tasks):
          remote_waits.append(remote_wait.acquire.remote())
          nested_waits.append(nested_wait.acquire.remote())
  
      [
          g.remote(remote_waits, nested_waits)
          for _ in range(num_root_tasks)
      ]
  
      errors = get_error_message(
          p, 1, ray_constants.WORKER_POOL_LARGE_ERROR
      )
  
      assert len(errors) == 1
      assert errors[0]["type"] == ray_constants.WORKER_POOL_LARGE_ERROR
      p.close()
  
imports: import os
  import signal
  import sys
  import threading
  import time
  import numpy
  import pytest
  import ray
  import ray._private.ray_constants
  import ray._private.utils
  from ray._private.ray_constants import DEBUG_AUTOSCALING_ERROR
  from ray._private.test_utils import Semaphore, get_error_message, get_log_batch, init_error_pubsub, run_string_as_driver_nonblocking, wait_for_condition
  from ray.cluster_utils import cluster_not_supported
  from ray.experimental.internal_kv import _internal_kv_get
label_desc: Use image node_id 2 for calling a global function with example usage: test_warning_for_too_many_nested_tasks(shutdown_only) and returns: int