code: def test_driver_exiting_when_worker_blocked(call_ray_start):
      # This test will create some drivers that submit some tasks and then
      # exit without waiting for the tasks to complete.
      address = call_ray_start
  
      ray.init(address=address)
  
      # Define a driver that creates two tasks, one that runs forever and the
      # other blocked on the first in a `ray.get`.
      driver_script = """
  import time
  import ray
  ray.init(address="{}")
  @ray.remote
  def f():
      time.sleep(10**6)
  @ray.remote
  def g():
      ray.get(f.remote())
  g.remote()
  time.sleep(1)
  print("success")
  """.format(
          address
      )
  
      # Create some drivers and let them exit and make sure everything is
      # still alive.
      for _ in range(3):
          out = run_string_as_driver(driver_script)
          # Make sure the first driver ran to completion.
          assert "success" in out
  
      # Define a driver that creates two tasks, one that runs forever and the
      # other blocked on the first in a `ray.wait`.
      driver_script = """
  import time
  import ray
  ray.init(address="{}")
  @ray.remote
  def f():
      time.sleep(10**6)
  @ray.remote
  def g():
      ray.wait([f.remote()])
  g.remote()
  time.sleep(1)
  print("success")
  """.format(
          address
      )
  
      # Create some drivers and let them exit and make sure everything is
      # still alive.
      for _ in range(3):
          out = run_string_as_driver(driver_script)
          # Make sure the first driver ran to completion.
          assert "success" in out
  
      # Define a driver that creates one task that depends on a nonexistent
      # object. This task will be queued as waiting to execute.
      driver_script_template = """
  import time
  import ray
  ray.init(address="{}")
  @ray.remote
  def g(x):
      return
  g.remote(ray.ObjectRef(ray._private.utils.hex_to_binary("{}")))
  time.sleep(1)
  print("success")
  """
  
      # Create some drivers and let them exit and make sure everything is
      # still alive.
      for _ in range(3):
          nonexistent_id = ray.ObjectRef.from_random()
          driver_script = driver_script_template.format(
              address, nonexistent_id.hex()
          )
          out = run_string_as_driver(driver_script)
          # Simulate the nonexistent dependency becoming available.
          ray._private.worker.global_worker.put_object(
              None, nonexistent_id
          )
          # Make sure the first driver ran to completion.
          assert "success" in out
  
      # Define a driver that calls `ray.wait` on a nonexistent object.
      driver_script_template = """
  import time
  import ray
  ray.init(address="{}")
  @ray.remote
  def g():
      ray.wait(ray.ObjectRef(ray._private.utils.hex_to_binary("{}")))
  g.remote()
  time.sleep(1)
  print("success")
  """
  
      # Create some drivers and let them exit and make sure everything is
      # still alive.
      for _ in range(3):
          nonexistent_id = ray.ObjectRef.from_random()
          driver_script = driver_script_template.format(
              address, nonexistent_id.hex()
          )
          out = run_string_as_driver(driver_script)
          # Simulate the nonexistent dependency becoming available.
          ray._private.worker.global_worker.put_object(
              None, nonexistent_id
          )
          # Make sure the first driver ran to completion.
          assert "success" in out
  
      @ray.remote
      def f():
          return 1
  
      # Make sure we can still talk with the raylet.
      ray.get(f.remote())
  
imports: import asyncio
  import os
  import subprocess
  import sys
  from pathlib import Path
  import psutil
  import pytest
  import ray
  import ray._private.ray_constants
  from ray._private.test_utils import Semaphore, check_call_ray, check_call_subprocess, kill_process_by_name, start_redis_instance, run_string_as_driver, run_string_as_driver_nonblocking, wait_for_children_of_pid, wait_for_children_of_pid_to_exit
  from ray._private.utils import detect_fate_sharing_support
label_desc: Use image node_id 6 for calling a global function with example usage: test_driver_exiting_when_worker_blocked(call_ray_start) and returns: int