class_docstr: A streaming Dataset executor.
  
  This implementation executes Dataset DAGs in a fully streamed way. It runs
  by setting up the operator topology, and then routing blocks through operators in
  a way that maximizes throughput under resource constraints.
class_docstr_tok: ["A","streaming","Dataset","executor",".","This","implementation","executes","Dataset","DAGs","in","a","fully","streamed","way",".","It","runs","by","setting","up","the","operator","topology",",","and","then","routing","blocks","through","operators","in","a","way","that","maximizes","throughput","under","resource","constraints","."]
docstr: Run one step of the scheduling loop.
  
  This runs a few general phases:
      1. Waiting for the next task completion using `ray.wait()`.
      2. Pulling completed refs into operator outqueues.
      3. Selecting and dispatching new inputs to operators.
  
  Returns:
      True if we should continue running the scheduling loop.
docstr_tok: ["Run","one","step","of","the","scheduling","loop",".","This","runs","a","few","general","phases",":","1",".","Waiting","for","the","next","task","completion","using","`","ray.wait","(",")","`",".","2",".","Pulling","completed","refs","into","operator","outqueues",".","3",".","Selecting","and","dispatching","new","inputs","to","operators",".","Returns",":","True","if","we","should","continue","running","the","scheduling","loop","."]
code: def _scheduling_loop_step(self, topology: Topology) -> bool:
      """Run one step of the scheduling loop.
  
      This runs a few general phases:
          1. Waiting for the next task completion using `ray.wait()`.
          2. Pulling completed refs into operator outqueues.
          3. Selecting and dispatching new inputs to operators.
  
      Returns:
          True if we should continue running the scheduling loop.
      """
  
      if DEBUG_TRACE_SCHEDULING:
          logger.get_logger().info("Scheduling loop step...")
  
      # Note: calling process_completed_tasks() is expensive since it incurs
      # ray.wait() overhead, so make sure to allow multiple dispatch per call for
      # greater parallelism.
      num_errored_blocks = process_completed_tasks(
          topology,
          self._backpressure_policies,
          self._max_errored_blocks,
      )
      if self._max_errored_blocks > 0:
          self._max_errored_blocks -= num_errored_blocks
      self._num_errored_blocks += num_errored_blocks
  
      # Dispatch as many operators as we can for completed tasks.
      limits = self._get_or_refresh_resource_limits()
      cur_usage = TopologyResourceUsage.of(topology)
      self._report_current_usage(cur_usage, limits)
      op = select_operator_to_run(
          topology,
          cur_usage,
          limits,
          self._backpressure_policies,
          ensure_at_least_one_running=self._consumer_idling(),
          execution_id=self._execution_id,
          autoscaling_state=self._autoscaling_state,
      )
      i = 0
      while op is not None:
          i += 1
          if i > PROGRESS_BAR_UPDATE_INTERVAL:
              break
          if DEBUG_TRACE_SCHEDULING:
              _debug_dump_topology(topology)
          topology[op].dispatch_next_task()
          cur_usage = TopologyResourceUsage.of(topology)
          op = select_operator_to_run(
              topology,
              cur_usage,
              limits,
              self._backpressure_policies,
              ensure_at_least_one_running=self._consumer_idling(),
              execution_id=self._execution_id,
              autoscaling_state=self._autoscaling_state,
          )
  
      update_operator_states(topology)
  
      # Update the progress bar to reflect scheduling decisions.
      for op_state in topology.values():
          op_state.refresh_progress_bar()
  
      self._update_stats_metrics(state="RUNNING")
      if (
          time.time() - self._last_debug_log_time
          >= DEBUG_LOG_INTERVAL_SECONDS
      ):
          _log_op_metrics(topology)
          if not DEBUG_TRACE_SCHEDULING:
              _debug_dump_topology(topology, log_to_stdout=False)
          self._last_debug_log_time = time.time()
  
      # Log metrics of newly completed operators.
      for op in topology:
          if op.completed() and not self._has_op_completed[op]:
              log_str = (
                  f"Operator {op} completed. "
                  f"Operator Metrics:\n{op._metrics.as_dict()}"
              )
              logger.get_logger(log_to_stdout=False).info(log_str)
              self._has_op_completed[op] = True
  
      # Keep going until all operators run to completion.
      return not all(op.completed() for op in topology)
  
imports: import os
  import threading
  import time
  import uuid
  from typing import Dict, Iterator, List, Optional
  from ray.data._internal.dataset_logger import DatasetLogger
  from ray.data._internal.execution.autoscaling_requester import get_or_create_autoscaling_requester_actor
  from ray.data._internal.execution.backpressure_policy import BackpressurePolicy, get_backpressure_policies
  from ray.data._internal.execution.interfaces import ExecutionOptions, ExecutionResources, Executor, OutputIterator, PhysicalOperator, RefBundle
  from ray.data._internal.execution.operators.input_data_buffer import InputDataBuffer
  from ray.data._internal.execution.streaming_executor_state import DEFAULT_OBJECT_STORE_MEMORY_LIMIT_FRACTION, AutoscalingState, OpState, Topology, TopologyResourceUsage, build_streaming_topology, process_completed_tasks, select_operator_to_run, update_operator_states
  from ray.data._internal.progress_bar import ProgressBar
  from ray.data._internal.stats import DatasetStats, StatsManager
  from ray.data._internal.util import cluster_resources
  from ray.data.context import DataContext
label_desc: Use image node_id 8 for calling the StreamingExecutor obj's underlying member method code with example usage: obj._scheduling_loop_step(topology) and returns: unknown