code: def test_generator_streaming_no_leak_upon_failures(
      monkeypatch, shutdown_only, crash_type
  ):
      with monkeypatch.context() as m:
          m.setenv(
              "RAY_testing_asio_delay_us",
              "CoreWorkerService.grpc_server.ReportGeneratorItemReturns=100000:1000000",
          )
          ray.init(num_cpus=1)
  
          @ray.remote
          def g():
              try:
                  gen = f.remote()
                  for ref in gen:
                      print(ref)
                      ray.get(ref)
              except Exception:
                  print("exception!")
                  del ref
  
              del gen
              gc.collect()
  
              # Only the ref g is alive.
              def verify():
                  print(list_objects())
                  return len(list_objects()) == 1
  
              wait_for_condition(verify)
              return True
  
          @ray.remote
          def f():
              for i in range(10):
                  time.sleep(0.2)
                  if i == 4:
                      if crash_type == "exception":
                          raise ValueError
                      else:
                          sys.exit(9)
                  yield 2
  
          for _ in range(5):
              ray.get(g.remote())
  
imports: import asyncio
  import pytest
  import numpy
  import sys
  import time
  import threading
  import gc
  from unittest.mock import patch, Mock
  import ray
  from ray._private.test_utils import wait_for_condition
  from ray.experimental.state.api import list_objects
  from ray._raylet import ObjectRefGenerator, ObjectRefStreamEndOfStreamError
  from ray.cloudpickle import dumps
  from ray.exceptions import WorkerCrashedError
label_desc: Use image node_id 6 for calling a global function with example usage: test_generator_streaming_no_leak_upon_failures(monkeypatch, shutdown_only, crash_type) and returns: True, unknown