code: def test_caller_failure_doesnt_hang(shutdown_only):
      """
      Verify if the caller fails (e.g., exception or crash),
      the generator is not backpressured forever.
      """
      ray.init(num_cpus=2)
  
      @ray.remote(_generator_backpressure_num_objects=1)
      def f():
          for i in range(5):
              print("yield", i)
              yield i
  
      @ray.remote
      class Caller:
          def f(self, failure_type, hang, task_name):
              gen = f.options(name=task_name).remote()
              assert ray.get(next(gen)) == 0
  
              if hang:
                  time.sleep(300)
  
              if failure_type == "exc":
                  raise ValueError
              elif failure_type == "exit":
                  sys.exit(1)
              else:
                  pass
  
      # If caller finishes, f should finish.
      print("Check caller finishes")
      caller = Caller.remote()
      caller.f.remote(None, False, "1")
  
      def verify():
          task = list_tasks(filters=[("name", "=", "1")])[0]
          assert task.state == "FINISHED"
          return True
  
      wait_for_condition(verify)
  
      # If caller fails with an exception, f should finish.
      print("Check caller raises an exception")
      caller = Caller.remote()
      r = caller.f.remote("exc", False, "2")  # noqa
  
      def verify():
          task = list_tasks(filters=[("name", "=", "2")])[0]
          assert task.state == "FINISHED"
          return True
  
      wait_for_condition(verify)
  
      # If caller fails with an exit, f (child) should finish.
      print("Check caller exits")
      caller = Caller.remote()
      r = caller.f.remote("exit", False, "3")  # noqa
  
      def verify():
          task = list_tasks(filters=[("name", "=", "3")])[0]
          assert task.state == "FINISHED"
          return True
  
      wait_for_condition(verify)
  
      # If caller fails by a system exit, f (child) should fail.
      print("Check caller killed")
      caller = Caller.remote()
      r = caller.f.remote(None, True, "4")  # noqa
      wait_for_condition(
          lambda: list_tasks(filters=[("name", "=", "4")])[0].state
          == "RUNNING"
      )
      ray.kill(caller)
  
      def verify():
          task = list_tasks(filters=[("name", "=", "4")])[0]
          assert task.state == "FAILED"
          return True
  
      wait_for_condition(verify)
  
file_path: ray/python/ray/tests/test_streaming_generator_backpressure.py
imports: import asyncio
  import pytest
  import numpy
  import sys
  import time
  import signal
  import os
  import ray
  from ray._private.test_utils import wait_for_condition
  from ray.util.state import list_tasks
label_desc: Use image node_id 3 for calling a global function with example usage: test_caller_failure_doesnt_hang(shutdown_only) and returns: True, True, True, True