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","."]
docstr: Blocking get operation.
  
  This is intended to be called concurrently from multiple clients.
docstr_tok: ["Blocking","get","operation",".","This","is","intended","to","be","called","concurrently","from","multiple","clients","."]
code: def get(
      self, epoch_id: int, output_split_idx: int
  ) -> Optional[Tuple[ObjectRef[Block], BlockMetadata]]:
      """Blocking get operation.
  
      This is intended to be called concurrently from multiple clients.
      """
  
      if epoch_id != self._cur_epoch:
          raise ValueError(
              "Invalid iterator: the dataset has moved on to another epoch."
          )
  
      try:
          # Ensure there is at least one bundle.
          with self._lock:
              if output_split_idx in self._next_bundle:
                  next_bundle = self._next_bundle[output_split_idx]
              else:
                  next_bundle = None
  
          # Fetch next bundle if needed.
          while next_bundle is None or not next_bundle.blocks:
              # This is a BLOCKING call, so do it outside the lock.
              next_bundle = self._output_iterator.get_next(
                  output_split_idx
              )
  
          block = next_bundle.blocks[-1]
          next_bundle = replace(
              next_bundle, blocks=next_bundle.blocks[:-1]
          )
  
          # Accumulate any remaining blocks in next_bundle map as needed.
          with self._lock:
              self._next_bundle[output_split_idx] = next_bundle
              if not next_bundle.blocks:
                  del self._next_bundle[output_split_idx]
  
          return block
      except StopIteration:
          return 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 4 for calling the SplitCoordinator obj's underlying member method code with example usage: obj.get(epoch_id, output_split_idx) and returns: block, None