class_docstr: Block prefetcher using ray.wait.
class_docstr_tok: ["Block","prefetcher","using","ray.wait","."]
code: def _run(self):
      while True:
          try:
              blocks_to_wait = []
              with self._condition:
                  if len(self._blocks) > 0:
                      blocks_to_wait, self._blocks = self._blocks[:], []
                  else:
                      if self._stopped:
                          return
                      blocks_to_wait = []
                      self._condition.wait()
              if len(blocks_to_wait) > 0:
                  ray.wait(
                      blocks_to_wait, num_returns=1, fetch_local=True
                  )
          except Exception:
              logger.exception("Error in prefetcher thread.")
  
imports: import logging
  import threading
  from contextlib import nullcontext
  from typing import Any, Callable, Iterator, List, Optional, Tuple
  import ray
  from ray.actor import ActorHandle
  from ray.data._internal.batcher import Batcher, ShufflingBatcher
  from ray.data._internal.block_batching.interfaces import Batch, BlockPrefetcher, CollatedBatch
  from ray.data._internal.stats import DatasetStats
  from ray.data.block import Block, BlockAccessor, DataBatch
  from ray.types import ObjectRef
  from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy
label_desc: Use image node_id 2 for calling the WaitBlockPrefetcher obj's underlying member method code with example usage: obj._run() without return types