code: def test_streaming_generator_load(shutdown_only):
      app = FastAPI()
  
      @serve.deployment(max_concurrent_queries=1000)
      @serve.ingress(app)
      class Router:
          def __init__(self, handle) -> None:
              self._h = handle.options(stream=True)
              self.total_recieved = 0
  
          @app.get("/")
          def stream_hi(self, request: Request) -> StreamingResponse:
              async def consume_obj_ref_gen():
                  obj_ref_gen = self._h.hi_gen.remote()
                  num_recieved = 0
                  async for chunk in obj_ref_gen:
                      num_recieved += 1
                      yield str(chunk.json())
  
              return StreamingResponse(
                  consume_obj_ref_gen(), media_type="text/plain"
              )
  
      @serve.deployment(max_concurrent_queries=1000)
      class SimpleGenerator:
          async def hi_gen(self):
              for i in range(100):
                  time.sleep(
                      0.001
                  )  # if change to async sleep, i don't see crash.
  
                  class Model(BaseModel):
                      msg = "a" * 56
  
                  yield Model()
  
      serve.run(Router.bind(SimpleGenerator.bind()))
  
      client_script = """
  import requests
  import time
  import io
  
  def send_serve_requests():
      request_meta = {
          "request_type": "InvokeEndpoint",
          "name": "Streamtest",
          "start_time": time.time(),
          "response_length": 0,
          "response": None,
          "context": {},
          "exception": None,
      }
      start_perf_counter = time.perf_counter()
      r = requests.get("http://localhost:8000", stream=True)
      print("status code: ", r.status_code)
      if r.status_code != 200:
          assert False
      else:
          for i, chunk in enumerate(r.iter_content(chunk_size=None, decode_unicode=True)):
              pass
          request_meta["response_time"] = (
              time.perf_counter() - start_perf_counter
          ) * 1000
          # events.request.fire(**request_meta)
  
  from concurrent.futures import ThreadPoolExecutor
  with ThreadPoolExecutor(max_workers=10) as executor:
      while True:
          futs = [executor.submit(send_serve_requests) for _ in range(100)]
          for f in futs:
              f.result()
  """
      for _ in range(5):
          print("submit a new clients!")
          proc = run_string_as_driver_nonblocking(client_script)
          # Wait sufficient time.
          time.sleep(5)
          proc.terminate()
          out_str = proc.stdout.read().decode("ascii")
          err_str = proc.stderr.read().decode("ascii")
          print(out_str, err_str)
          for actor in list_actors():
              assert actor.state != "DEAD"
  
imports: import asyncio
  import pytest
  import numpy
  import sys
  import time
  from collections import Counter
  from starlette.responses import StreamingResponse
  from starlette.requests import Request
  from fastapi import FastAPI
  from ray import serve
  from pydantic import BaseModel
  import ray
  from ray._raylet import ObjectRefGenerator
  from ray.exceptions import WorkerCrashedError
  from ray._private.test_utils import run_string_as_driver_nonblocking
  from ray.util.state import list_actors
label_desc: Use image node_id 6 for calling a global function with example usage: test_streaming_generator_load(shutdown_only) and returns: StreamingResponse