class_docstr: Coordinator actor for routing blocks to output splits.
  
  This actor runs a streaming executor locally on its main thread. Clients can
  retrieve results via actor calls running on other threads.
class_docstr_tok: ["Coordinator","actor","for","routing","blocks","to","output","splits",".","This","actor","runs","a","streaming","executor","locally","on","its","main","thread",".","Clients","can","retrieve","results","via","actor","calls","running","on","other","threads","."]
code: def __init__(
      self,
      dataset: "Dataset",
      n: int,
      equal: bool,
      locality_hints: Optional[List[NodeIdStr]],
  ):
      # Automatically set locality with output to the specified location hints.
      if locality_hints:
          dataset.context.execution_options.locality_with_output = (
              locality_hints
          )
          logger.info(
              f"Auto configuring locality_with_output={locality_hints}"
          )
  
      # Set current DataContext.
      ray.data.DataContext._set_current(dataset.context)
  
      self._base_dataset = dataset
      self._n = n
      self._equal = equal
      self._locality_hints = locality_hints
      self._lock = threading.RLock()
      self._executor = None
  
      # Guarded by self._lock.
      self._next_bundle: Dict[int, RefBundle] = {}
      self._unfinished_clients_in_epoch = n
      self._cur_epoch = -1
  
      def gen_epochs():
          while True:
              executor = StreamingExecutor(
                  copy.deepcopy(dataset.context.execution_options),
                  create_dataset_tag(
                      self._base_dataset._name, self._base_dataset._uuid
                  ),
              )
              self._executor = executor
  
              def add_split_op(dag):
                  return OutputSplitter(dag, n, equal, locality_hints)
  
              output_iterator = execute_to_legacy_bundle_iterator(
                  executor,
                  dataset._plan,
                  True,
                  dataset._plan._dataset_uuid,
                  dag_rewrite=add_split_op,
              )
              yield output_iterator
  
      self._next_epoch = gen_epochs()
      self._output_iterator = None
  
imports: import copy
  import logging
  import threading
  import time
  from dataclasses import replace
  from typing import TYPE_CHECKING, Dict, Iterator, List, Optional, Tuple, Union
  import ray
  from ray.data._internal.execution.interfaces import NodeIdStr, RefBundle
  from ray.data._internal.execution.legacy_compat import execute_to_legacy_bundle_iterator
  from ray.data._internal.execution.operators.output_splitter import OutputSplitter
  from ray.data._internal.execution.streaming_executor import StreamingExecutor
  from ray.data._internal.stats import DatasetStats, DatasetStatsSummary
  from ray.data._internal.util import create_dataset_tag
  from ray.data.block import Block, BlockMetadata
  from ray.data.iterator import DataIterator
  from ray.types import ObjectRef
  from ray.util.debug import log_once
  from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy
label_desc: Use image node_id 1 to create a new SplitCoordinator object  with example: obj = SplitCoordinator(dataset, n, equal, locality_hints)