code: def test_drivers_release_resources(call_ray_start):
      address = call_ray_start
  
      # Define a driver that creates an actor and exits.
      driver_script1 = """
  import time
  import ray
  
  ray.init(address="{}")
  
  @ray.remote
  def f(duration):
      time.sleep(duration)
  
  @ray.remote(num_gpus=1)
  def g(duration):
      time.sleep(duration)
  
  @ray.remote(num_gpus=1)
  class Foo:
      def __init__(self):
          pass
  
  # Make sure some resources are available for us to run tasks.
  ray.get(f.remote(0))
  ray.get(g.remote(0))
  
  # Start a bunch of actors and tasks that use resources. These should all be
  # cleaned up when this driver exits.
  foos = [Foo.remote() for _ in range(100)]
  [f.remote(10 ** 6) for _ in range(100)]
  
  print("success")
  """.format(
          address
      )
  
      driver_script2 = (
          driver_script1
          + "import sys\nsys.stdout.flush()\ntime.sleep(10 ** 6)\n"
      )
  
      def wait_for_success_output(process_handle, timeout=10):
          # Wait until the process prints "success" and then return.
          start_time = time.time()
          while time.time() - start_time < timeout:
              output_line = ray._private.utils.decode(
                  process_handle.stdout.readline()
              ).strip()
              print(output_line)
              if output_line == "success":
                  return
              time.sleep(1)
          raise RayTestTimeoutException(
              "Timed out waiting for process to print success."
          )
  
      # Make sure we can run this driver repeatedly, which means that resources
      # are getting released in between.
      for _ in range(5):
          out = run_string_as_driver(driver_script1)
          # Make sure the first driver ran to completion.
          assert "success" in out
          # Also make sure that this works when the driver exits ungracefully.
          process_handle = run_string_as_driver_nonblocking(
              driver_script2
          )
          wait_for_success_output(process_handle)
          # Kill the process ungracefully.
          process_handle.kill()
  
file_path: ray/python/ray/tests/test_multi_node.py
imports: import os
  import sys
  import time
  import psutil
  import pytest
  import ray
  from ray._private import ray_constants
  from ray._private.test_utils import RayTestTimeoutException, get_error_message, init_error_pubsub, object_memory_usage, run_string_as_driver, run_string_as_driver_nonblocking, wait_for_condition
label_desc: Use image node_id 7 for calling a global function with example usage: test_drivers_release_resources(call_ray_start) without return types