code: def test_dynamic_generator(
      ray_start_regular, use_actors, store_in_plasma, num_returns_type
  ):
      if not use_actors:
  
          @ray.remote(num_returns=num_returns_type)
          def dynamic_generator(num_returns, store_in_plasma):
              for i in range(num_returns):
                  if store_in_plasma:
                      yield np.ones(1_000_000, dtype=np.int8) * i
                  else:
                      yield [i]
  
          remote_generator_fn = dynamic_generator
      else:
  
          @ray.remote
          class Generator:
              def __init__(self):
                  pass
  
              def generator(self, num_returns, store_in_plasma):
                  for i in range(num_returns):
                      if store_in_plasma:
                          yield np.ones(1_000_000, dtype=np.int8) * i
                      else:
                          yield [i]
  
          g = Generator.remote()
          remote_generator_fn = g.generator
  
      @ray.remote
      def read(gen):
          for i, ref in enumerate(gen):
              if ray.get(ref)[0] != i:
                  return False
          return True
  
      gen = ray.get(
          remote_generator_fn.options(
              num_returns=num_returns_type
          ).remote(10, store_in_plasma)
      )
      for i, ref in enumerate(gen):
          assert ray.get(ref)[0] == i
  
      # Test empty generator.
      gen = ray.get(
          remote_generator_fn.options(
              num_returns=num_returns_type
          ).remote(0, store_in_plasma)
      )
      assert len(list(gen)) == 0
  
      # Check that passing as task arg.
      if num_returns_type == "dynamic":
          gen = remote_generator_fn.options(
              num_returns=num_returns_type
          ).remote(10, store_in_plasma)
          assert ray.get(read.remote(gen))
          assert ray.get(read.remote(ray.get(gen)))
      else:
          with pytest.raises(TypeError):
              gen = remote_generator_fn.options(
                  num_returns=num_returns_type
              ).remote(10, store_in_plasma)
              assert ray.get(read.remote(gen))
  
      # Also works if we override num_returns with a static value.
      ray.get(
          read.remote(
              remote_generator_fn.options(num_returns=10).remote(
                  10, store_in_plasma
              )
          )
      )
  
      if num_returns_type == "dynamic":
          # Normal remote functions don't work with num_returns="dynamic".
          @ray.remote(num_returns=num_returns_type)
          def static(num_returns):
              return list(range(num_returns))
  
          with pytest.raises(ray.exceptions.RayTaskError):
              gen = ray.get(static.remote(3))
              for ref in gen:
                  ray.get(ref)
  
imports: import pytest
  import numpy
  import sys
  import time
  import gc
  from unittest.mock import Mock
  import ray
  from ray.util.client.ray_client_helpers import ray_start_client_server_for_address
  from ray._private.client_mode_hook import enable_client_mode
  from ray.tests.conftest import call_ray_start_context
label_desc: Use image node_id 6 for calling a global function with example usage: test_dynamic_generator(ray_start_regular, use_actors, store_in_plasma, num_returns_type) and returns: True, list, False