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: Return concrete limits for use at the current time.
  
  This method autodetects any unspecified execution resource limits based on the
  current cluster size, refreshing these values periodically to support cluster
  autoscaling.
docstr_tok: ["Return","concrete","limits","for","use","at","the","current","time",".","This","method","autodetects","any","unspecified","execution","resource","limits","based","on","the","current","cluster","size",",","refreshing","these","values","periodically","to","support","cluster","autoscaling","."]
code: def _get_or_refresh_resource_limits(self) -> ExecutionResources:
      """Return concrete limits for use at the current time.
  
      This method autodetects any unspecified execution resource limits based on the
      current cluster size, refreshing these values periodically to support cluster
      autoscaling.
      """
      base = self._options.resource_limits
      exclude = self._options.exclude_resources
      cluster = cluster_resources()
  
      cpu = base.cpu
      if cpu is None:
          cpu = cluster.get("CPU", 0.0) - (exclude.cpu or 0.0)
      gpu = base.gpu
      if gpu is None:
          gpu = cluster.get("GPU", 0.0) - (exclude.gpu or 0.0)
      object_store_memory = base.object_store_memory
      if object_store_memory is None:
          object_store_memory = round(
              DEFAULT_OBJECT_STORE_MEMORY_LIMIT_FRACTION
              * cluster.get("object_store_memory", 0.0)
          ) - (exclude.object_store_memory or 0)
  
      return ExecutionResources(
          cpu=cpu,
          gpu=gpu,
          object_store_memory=object_store_memory,
      )
  
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 10 for calling the StreamingExecutor obj's underlying member method code with example usage: obj._get_or_refresh_resource_limits() and returns: ExecutionResources