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 _print_warning(self, op: "PhysicalOperator", idle_time: float):
      if self._warning_printed:
          return
      self._warning_printed = True
      msg = (
          f"Operator {op} is running but has no outputs for {idle_time} seconds."
          " Execution may be slower than expected.\n"
          "Ignore this warning if your UDF is expected to be slow."
          " Otherwise, this can happen when there are fewer cluster resources"
          " available to Ray Data than expected."
          " If you have non-Data tasks or actors running in the cluster, exclude"
          " their resources from Ray Data with"
          " `DataContext.get_current().execution_options.exclude_resources`."
          " This message will only print once."
      )
      logger.get_logger().warning(msg)
  
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 3 for calling the StreamingOutputBackpressurePolicy obj's underlying member method code with example usage: obj._print_warning(op, idle_time) without return types