code: def test_actor_streaming_generator(shutdown_only, store_in_plasma):
      """Test actor/async actor with sync/async generator interfaces."""
      ray.init()
  
      @ray.remote
      class Actor:
          def f(self, ref):
              for i in range(3):
                  yield i
  
          async def async_f(self, ref):
              for i in range(3):
                  await asyncio.sleep(0.1)
                  yield i
  
          def g(self):
              return 3
  
      a = Actor.remote()
      if store_in_plasma:
          arr = np.random.rand(5 * 1024 * 1024)
      else:
          arr = 3
  
      def verify_sync_task_executor():
          generator = a.f.remote(ray.put(arr))
          # Verify it works with next.
          assert isinstance(generator, ObjectRefGenerator)
          assert ray.get(next(generator)) == 0
          assert ray.get(next(generator)) == 1
          assert ray.get(next(generator)) == 2
          with pytest.raises(StopIteration):
              ray.get(next(generator))
  
          # Verify it works with for.
          generator = a.f.remote(ray.put(3))
          for index, ref in enumerate(generator):
              assert index == ray.get(ref)
  
      def verify_async_task_executor():
          # Verify it works with next.
          generator = a.async_f.remote(ray.put(arr))
          assert isinstance(generator, ObjectRefGenerator)
          assert ray.get(next(generator)) == 0
          assert ray.get(next(generator)) == 1
          assert ray.get(next(generator)) == 2
  
          # Verify it works with for.
          generator = a.f.remote(ray.put(3))
          for index, ref in enumerate(generator):
              assert index == ray.get(ref)
  
      async def verify_sync_task_async_generator():
          # Verify anext
          async_generator = a.f.remote(ray.put(arr))
          assert isinstance(async_generator, ObjectRefGenerator)
          for expected in range(3):
              ref = await async_generator.__anext__()
              assert await ref == expected
          with pytest.raises(StopAsyncIteration):
              await async_generator.__anext__()
  
          # Verify async for.
          async_generator = a.f.remote(ray.put(arr))
          expected = 0
          async for ref in async_generator:
              value = await ref
              assert expected == value
              expected += 1
  
      async def verify_async_task_async_generator():
          async_generator = a.async_f.remote(ray.put(arr))
          assert isinstance(async_generator, ObjectRefGenerator)
          for expected in range(3):
              ref = await async_generator.__anext__()
              assert await ref == expected
          with pytest.raises(StopAsyncIteration):
              await async_generator.__anext__()
  
          # Verify async for.
          async_generator = a.async_f.remote(ray.put(arr))
          expected = 0
          async for ref in async_generator:
              value = await ref
              assert expected == value
              expected += 1
  
      verify_sync_task_executor()
      verify_async_task_executor()
      asyncio.run(verify_sync_task_async_generator())
      asyncio.run(verify_async_task_async_generator())
  
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 10 for calling a global function with example usage: test_actor_streaming_generator(shutdown_only, store_in_plasma) and returns: int