class_docstr: A backpressure policy that throttles the streaming outputs of the `DataOpTask`s.
  
  The are 2 levels of configs to control the behavior:
  - At the Ray Core level, we use
    `MAX_BLOCKS_IN_GENERATOR_BUFFER` to limit the number of blocks buffered in
    the streaming generator of each OpDataTask. When it's reached, the task will
    be blocked at `yield` until the caller reads another `ObjectRef.
  - At the Ray Data level, we use
    `MAX_BLOCKS_IN_OP_OUTPUT_QUEUE` to limit the number of blocks buffered in the
    output queue of each operator. When it's reached, we'll stop reading from the
    streaming generators of the op's tasks, and thus trigger backpressure at the
    Ray Core level.
  
  Thus, total number of buffered blocks for each operator can be
  `MAX_BLOCKS_IN_GENERATOR_BUFFER * num_running_tasks +
  MAX_BLOCKS_IN_OP_OUTPUT_QUEUE`.
class_docstr_tok: ["A","backpressure","policy","that","throttles","the","streaming","outputs","of","the","`","DataOpTask","`","s",".","The","are","2","levels","of","configs","to","control","the","behavior",":","-","At","the","Ray","Core","level",",","we","use","`","MAX_BLOCKS_IN_GENERATOR_BUFFER","`","to","limit","the","number","of","blocks","buffered","in","the","streaming","generator","of","each","OpDataTask",".","When","it","'s","reached",",","the","task","will","be","blocked","at","`","yield","`","until","the","caller","reads","another","`","ObjectRef",".","-","At","the","Ray","Data","level",",","we","use","`","MAX_BLOCKS_IN_OP_OUTPUT_QUEUE","`","to","limit","the","number","of","blocks","buffered","in","the","output","queue","of","each","operator",".","When","it","'s","reached",",","we","'ll","stop","reading","from","the","streaming","generators","of","the","op","'s","tasks",",","and","thus","trigger","backpressure","at","the","Ray","Core","level",".","Thus",",","total","number","of","buffered","blocks","for","each","operator","can","be","`","MAX_BLOCKS_IN_GENERATOR_BUFFER","*","num_running_tasks","+","MAX_BLOCKS_IN_OP_OUTPUT_QUEUE","`","."]
code: def __init__(self, topology: "Topology"):
      data_context = ray.data.DataContext.get_current()
      self._max_num_blocks_in_streaming_gen_buffer = (
          data_context.get_config(
              self.MAX_BLOCKS_IN_GENERATOR_BUFFER_CONFIG_KEY,
              self.MAX_BLOCKS_IN_GENERATOR_BUFFER,
          )
      )
      assert self._max_num_blocks_in_streaming_gen_buffer > 0
      # The `_generator_backpressure_num_objects` parameter should be
      # `2 * self._max_num_blocks_in_streaming_gen_buffer` because we yield
      # 2 objects for each block: the block and the block metadata.
      data_context._task_pool_data_task_remote_args[
          "_generator_backpressure_num_objects"
      ] = (2 * self._max_num_blocks_in_streaming_gen_buffer)
  
      self._max_num_blocks_in_op_output_queue = data_context.get_config(
          self.MAX_BLOCKS_IN_OP_OUTPUT_QUEUE_CONFIG_KEY,
          self.MAX_BLOCKS_IN_OP_OUTPUT_QUEUE,
      )
      assert self._max_num_blocks_in_op_output_queue > 0
  
      # Latest number of outputs and the last time when the number changed
      # for each op.
      self._last_num_outputs_and_time: Dict[
          "PhysicalOperator", Tuple[int, float]
      ] = defaultdict(lambda: (0, time.time()))
      self._warning_printed = False
  
file_path: ray/python/ray/data/_internal/execution/backpressure_policy/streaming_output_backpressure_policy.py
imports: import time
  from collections import defaultdict
  from typing import TYPE_CHECKING, Dict, Tuple
  import ray
  from .backpressure_policy import BackpressurePolicy
  from ray.data._internal.dataset_logger import DatasetLogger
label_desc: Use image node_id 1 to create a new StreamingOutputBackpressurePolicy object from inherited base classes: BackpressurePolicy with example: obj = StreamingOutputBackpressurePolicy(topology)