code: def test_ray_datasetlike_mini_stress_test(
      monkeypatch,
      ray_start_cluster,
      backpressure,
      delay_latency,
      threshold,
  ):
      """
      Test a workload that's like ray dataset + lineage reconstruction.
      """
      if not backpressure:
          if delay_latency == 0.1 and threshold == 1:
              return
          elif delay_latency == 1:
              return
  
      with monkeypatch.context() as m:
          m.setenv(
              "RAY_testing_asio_delay_us",
              "CoreWorkerService.grpc_server."
              "ReportGeneratorItemReturns=10000:1000000",
          )
          cluster = ray_start_cluster
          cluster.add_node(
              num_cpus=1,
              resources={"head": 1},
              _system_config=RECONSTRUCTION_CONFIG,
              enable_object_reconstruction=True,
          )
          ray.init(address=cluster.address)
  
          if backpressure:
              threshold = 1
          else:
              threshold = -1
  
          @ray.remote(
              max_retries=-1,
              _generator_backpressure_num_objects=threshold,
          )
          def dynamic_generator(num_returns):
              for i in range(num_returns):
                  time.sleep(0.1)
                  yield np.ones(1_000_000, dtype=np.int8) * i
  
          @ray.remote(num_cpus=0, resources={"head": 1})
          def driver():
              unready = [dynamic_generator.remote(10) for _ in range(5)]
              ready = []
              while unready:
                  for a in unready:
                      print(a._generator_ref)
                  ready, unready = ray.wait(
                      unready, num_returns=len(unready), timeout=0.1
                  )
                  for r in ready:
                      try:
                          ref = next(r)
                          print(ref)
                          ray.get(ref)
                      except StopIteration:
                          pass
                      else:
                          unready.append(r)
              return None
  
          ref = driver.remote()
  
          nodes = []
          for _ in range(4):
              nodes.append(
                  cluster.add_node(
                      num_cpus=1, object_store_memory=10**8
                  )
              )
          cluster.wait_for_nodes()
  
          for _ in range(10):
              time.sleep(0.1)
              node_to_kill = random.choices(nodes)[0]
              nodes.remove(node_to_kill)
              cluster.remove_node(node_to_kill, allow_graceful=False)
              nodes.append(
                  cluster.add_node(
                      num_cpus=1, object_store_memory=10**8
                  )
              )
  
          ray.get(ref)
          del ref
  
          assert_no_leak()
  
file_path: ray/python/ray/tests/test_streaming_generator_4.py
imports: import pytest
  import numpy
  import sys
  import time
  import gc
  import random
  import asyncio
  from typing import Optional
  from pydantic import BaseModel
  import ray
label_desc: Use image node_id 2 for calling a global function with example usage: test_ray_datasetlike_mini_stress_test(monkeypatch, ray_start_cluster, backpressure, delay_latency, threshold) and returns: None