class_docstr: Snapshot of resource usage in a `Topology` object.
  
  The stats here can be computed on the fly from any `Topology`; this class
  serves only a convenience wrapper to access the current usage snapshot.
docstr: Calculate the resource usage of the given topology.
docstr_tok: ["Calculate","the","resource","usage","of","the","given","topology","."]
code: def of(topology: Topology) -> "TopologyResourceUsage":
      """Calculate the resource usage of the given topology."""
      downstream_usage = {}
      cur_usage = ExecutionResources(0, 0, 0)
      # Iterate from last to first operator.
      for op, state in reversed(topology.items()):
          cur_usage = cur_usage.add(op.current_resource_usage())
          # Don't count input refs towards dynamic memory usage, as they have been
          # pre-created already outside this execution.
          if not isinstance(op, InputDataBuffer):
              cur_usage.object_store_memory += (
                  state.outqueue_memory_usage()
              )
          # Subtract one from denom to account for input buffer.
          f = (1.0 + len(downstream_usage)) / max(
              1.0, len(topology) - 1.0
          )
          downstream_usage[op] = DownstreamMemoryInfo(
              topology_fraction=min(1.0, f),
              object_store_memory=cur_usage.object_store_memory,
          )
      return TopologyResourceUsage(cur_usage, downstream_usage)
  
imports: import math
  import threading
  import time
  from collections import defaultdict, deque
  from dataclasses import dataclass
  from typing import Dict, List, Optional, Tuple
  import ray
  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
  from ray.data._internal.execution.interfaces import ExecutionOptions, ExecutionResources, PhysicalOperator, RefBundle
  from ray.data._internal.execution.interfaces.physical_operator import DataOpTask, MetadataOpTask, OpTask, Waitable
  from ray.data._internal.execution.operators.base_physical_operator import AllToAllOperator
  from ray.data._internal.execution.operators.input_data_buffer import InputDataBuffer
  from ray.data._internal.execution.util import memory_string
  from ray.data._internal.progress_bar import ProgressBar
  from ray.data.context import DataContext
label_desc: Use image node_id 1 for calling the TopologyResourceUsage obj's underlying member method code with example usage: obj.of(topology) and returns: TopologyResourceUsage