code: def test_autoshutdown_dangling_executors(ray_start_10_cpus_shared):
      from ray.data._internal.execution import streaming_executor
  
      num_runs = 5
  
      # Test that when an interator is fully consumed,
      # the executor should be shut down.
      initial = streaming_executor._num_shutdown
      for _ in range(num_runs):
          ds = ray.data.range(100).repartition(10)
          it = iter(ds.iter_batches(batch_size=10, prefetch_batches=0))
          while True:
              try:
                  next(it)
              except StopIteration:
                  break
      assert streaming_executor._num_shutdown - initial == num_runs
  
      # Test that when an partially-consumed iterator is deleted,
      # the executor should be shut down.
      initial = streaming_executor._num_shutdown
      for _ in range(num_runs):
          ds = ray.data.range(100).repartition(10)
          it = iter(ds.iter_batches(batch_size=10, prefetch_batches=0))
          next(it)
          del it
          del ds
      assert streaming_executor._num_shutdown - initial == num_runs
  
      # Test that the executor is shut down when it's deleted,
      # even if not using iterators.
      initial = streaming_executor._num_shutdown
      for _ in range(num_runs):
          executor = StreamingExecutor(ExecutionOptions())
          o = InputDataBuffer([])
          # Start the executor. Because non-started executors don't
          # need to be shut down.
          executor.execute(o)
          del executor
      assert streaming_executor._num_shutdown - initial == num_runs
  
imports: import itertools
  import random
  import threading
  import time
  from typing import Any, List
  import pandas
  import pytest
  import ray
  from ray import cloudpickle
  from ray._private.test_utils import wait_for_condition
  from ray.data._internal.execution.interfaces import ExecutionOptions, ExecutionResources, RefBundle
  from ray.data._internal.execution.operators.base_physical_operator import AllToAllOperator
  from ray.data._internal.execution.operators.input_data_buffer import InputDataBuffer
  from ray.data._internal.execution.operators.map_operator import MapOperator
  from ray.data._internal.execution.operators.map_transformer import create_map_transformer_from_block_fn
  from ray.data._internal.execution.operators.output_splitter import OutputSplitter
  from ray.data._internal.execution.streaming_executor import StreamingExecutor
  from ray.data._internal.execution.util import make_ref_bundles
  from ray.data.context import DataContext
  from ray.data.tests.conftest import *
  from ray.data.tests.util import extract_values
label_desc: Use image node_id 3 for calling a global function with example usage: test_autoshutdown_dangling_executors(ray_start_10_cpus_shared) without return types