code: def test_no_memory_store_obj_leak(shutdown_only):
      """Fixes https://github.com/ray-project/ray/issues/38089
  
      Verify there's no leak from in-memory object store when
      using a streaming generator.
      """
      ray.init()
  
      @ray.remote
      def f():
          for _ in range(10):
              yield 1
  
      for _ in range(10):
          for ref in f.remote():
              del ref
  
          time.sleep(0.2)
  
      core_worker = ray._private.worker.global_worker.core_worker
      assert core_worker.get_memory_store_size() == 0
      assert_no_leak()
  
      for _ in range(10):
          for ref in f.remote():
              break
  
          time.sleep(0.2)
  
      del ref
      core_worker = ray._private.worker.global_worker.core_worker
      assert core_worker.get_memory_store_size() == 0
      assert_no_leak()
  
file_path: ray/python/ray/tests/test_streaming_generator_2.py
imports: import asyncio
  import pytest
  import numpy
  import sys
  import time
  import gc
  import ray
  from ray.experimental.state.api import list_actors
label_desc: Use image node_id 8 for calling a global function with example usage: test_no_memory_store_obj_leak(shutdown_only) without return types