code: def test_worker_capping_run_chained_tasks(shutdown_only):
      ray.init(num_cpus=2)
  
      @ray.remote(num_cpus=0.5)
      def foo(x):
          if x > 1:
              return ray.get(foo.remote(x - 1)) + x
          else:
              time.sleep(5)
              return x
  
      # Run a chain of tasks which exceed `num_cpus` in amount, but the CPU
      # resource requirement is still within `num_cpus`.
      obj = foo.remote(4)
      wait_for_condition(lambda: len(get_workers()) == 4)
  
      ray.get(obj)
      # After finished the tasks, some workers are killed to keep the total
      # number of workers <= num_cpus.
      wait_for_condition(lambda: len(get_workers()) == 2)
  
      time.sleep(1)
      # The two remaining workers stay alive forever.
      assert len(get_workers()) == 2
  
file_path: ray/python/ray/tests/test_multi_tenancy.py
imports: import os
  import sys
  import time
  import pytest
  import numpy
  import ray
  from ray.core.generated import common_pb2
  from ray.core.generated import node_manager_pb2, node_manager_pb2_grpc
  from ray._private.test_utils import wait_for_condition, run_string_as_driver, run_string_as_driver_nonblocking
  from ray._private.utils import init_grpc_channel
label_desc: Use image node_id 7 for calling a global function with example usage: test_worker_capping_run_chained_tasks(shutdown_only) and returns: unknown, x