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 calculate_max_blocks_to_read_per_op(
      self, topology: "Topology"
  ) -> Dict["OpState", int]:
      max_blocks_to_read_per_op: Dict["OpState", int] = {}
  
      # Indicates if the immediate downstream operator is idle.
      downstream_idle = False
  
      for op, state in reversed(topology.items()):
          max_blocks_to_read_per_op[state] = (
              self._max_num_blocks_in_op_output_queue
              - state.outqueue_num_blocks()
          )
  
          if downstream_idle:
              max_blocks_to_read_per_op[state] = max(
                  max_blocks_to_read_per_op[state],
                  1,
              )
  
          # An operator is considered idle if either of the following is true:
          # - It has no active tasks.
          #   - This can happen when all resources are used by upstream operators.
          # - It has active tasks, but no outputs for at least
          #   `MAX_OUTPUT_IDLE_SECONDS`.
          #   - This can happen when non-Data code preempted cluster resources, and
          #   - some of the active tasks don't actually have enough resources to run.
          #
          # If the operator is idle, we'll temporarily unblock backpressure by
          # allowing reading at least one block from its upstream
          # to avoid deadlock.
          # NOTE, these 2 conditions don't necessarily mean deadlock.
          # The first case can also happen when the upstream operator hasn't outputted
          # any blocks yet. While the second case can also happen when the task is
          # expected to output data slowly.
          # The false postive cases are fine as we only allow reading one block
          # each time.
          downstream_idle = False
          if op.num_active_tasks() == 0:
              downstream_idle = True
          else:
              cur_num_outputs = state.op.metrics.num_outputs_generated
              cur_time = time.time()
              (
                  last_num_outputs,
                  last_time,
              ) = self._last_num_outputs_and_time[state.op]
              if cur_num_outputs > last_num_outputs:
                  self._last_num_outputs_and_time[state.op] = (
                      cur_num_outputs,
                      cur_time,
                  )
              else:
                  if (
                      cur_time - last_time
                      > self.MAX_OUTPUT_IDLE_SECONDS
                  ):
                      downstream_idle = True
                      self._print_warning(
                          state.op, cur_time - last_time
                      )
      return max_blocks_to_read_per_op
  
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 2 for calling the StreamingOutputBackpressurePolicy obj's underlying member method code with example usage: obj.calculate_max_blocks_to_read_per_op(topology) and returns: max_blocks_to_read_per_op