code: def test_streaming_generator_backpressure_basic(shutdown_only, actor):
      """Verify backpressure works with
      _generator_backpressure_num_objects = 0
      """
      ray.init(num_cpus=1)
  
      @ray.remote
      class Reporter:
          def __init__(self):
              self.reported = set()
  
          def report(self, i):
              self.reported.add(i)
  
          def reported(self):
              return self.reported
  
      TOTAL_RETURN = 10
  
      @ray.remote(_generator_backpressure_num_objects=1)
      def f(reporter):
          for i in range(TOTAL_RETURN):
              print("yield ", i)
              ray.get(reporter.report.remote(i))
              yield i
  
      @ray.remote
      class A:
          @ray.method(_generator_backpressure_num_objects=1)
          def f(self, reporter):
              for i in range(TOTAL_RETURN):
                  print("yield ", i)
                  ray.get(reporter.report.remote(i))
                  yield i
  
      reporter = Reporter.remote()
  
      def check_reported(i):
          return i in ray.get(reporter.reported.remote())
  
      if actor:
          a = A.remote()
          gen = a.f.remote(reporter)
      else:
          gen = f.remote(reporter)
  
      # Don't iterate everything to test is deleing the generator
      # would stop backpressure (see the end of the test).
      for i in range(TOTAL_RETURN - 4):
          print("iteration ", i)
          r, _ = ray.wait([gen])
          assert len(r) == 1
          wait_for_condition(lambda: check_reported(i))
          wait_for_condition(lambda: not check_reported(i + 1))
          # Wait a little bit to make sure it is backpressured.
          time.sleep(2)
          wait_for_condition(lambda: not check_reported(i + 1))
          # Consume the ref -> task will progress.
          ray.get(next(gen))
          wait_for_condition(lambda: check_reported(i + 1))
  
      """
      Verify deleting a generator will stop backpressure
      and proceed a task.
      """
  
      del gen
      del r
      wait_for_condition(lambda: check_reported(TOTAL_RETURN - 1))
  
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 1 for calling a global function with example usage: test_streaming_generator_backpressure_basic(shutdown_only, actor) and returns: unknown, self