code: def test_threaded_actor_generator_backpressure(shutdown_only):
      ray.init()
  
      @ray.remote(max_concurrency=10)
      class Actor:
          def f(self):
              for i in range(30):
                  time.sleep(0.1)
                  print("yield", i)
                  yield np.ones(1024 * 1024) * i
  
      async def main():
          a = Actor.remote()
  
          async def run():
              i = 0
              gen = a.f.options(
                  _generator_backpressure_num_objects=1
              ).remote()
              async for ref in gen:
                  val = ray.get(ref)
                  print(val)
                  print(ref)
                  assert np.array_equal(val, np.ones(1024 * 1024) * i)
                  i += 1
                  del ref
  
          coroutines = [run() for _ in range(10)]
  
          await asyncio.gather(*coroutines)
  
      asyncio.run(main())
  
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 5 for calling a global function with example usage: test_threaded_actor_generator_backpressure(shutdown_only) without return types