code: def test_generator_dist_chain(ray_start_cluster):
      """E2E test to verify chain of generator works properly."""
      cluster = ray_start_cluster
      cluster.add_node(
          num_cpus=0, object_store_memory=1 * 1024 * 1024 * 1024
      )
      ray.init()
      cluster.add_node(num_cpus=1)
      cluster.add_node(num_cpus=1)
      cluster.add_node(num_cpus=1)
      cluster.add_node(num_cpus=1)
  
      @ray.remote
      class ChainActor:
          def __init__(self, child=None):
              self.child = child
  
          def get_data(self):
              if not self.child:
                  for _ in range(10):
                      time.sleep(0.1)
                      yield np.ones(5 * 1024 * 1024)
              else:
                  for data in self.child.get_data.remote():
                      yield ray.get(data)
  
      chain_actor = ChainActor.remote()
      chain_actor_2 = ChainActor.remote(chain_actor)
      chain_actor_3 = ChainActor.remote(chain_actor_2)
      chain_actor_4 = ChainActor.remote(chain_actor_3)
  
      for ref in chain_actor_4.get_data.remote():
          assert np.array_equal(np.ones(5 * 1024 * 1024), ray.get(ref))
          print("getting the next data")
          del ref
  
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 8 for calling a global function with example usage: test_generator_dist_chain(ray_start_cluster) without return types