code: def _get_dp_sharding_perf(
      batch_sizes: List[int],
      world_size: int,
      local_world_size: int,
      input_lengths: List[float],
      grad_num_elem: int,
      emb_dim: int,
      input_data_type_size: float,
      table_data_type_size: float,
      num_poolings: List[float],
      device_bw: float,
      inter_host_bw: float,
      bwd_compute_multiplier: float,
      is_pooled: bool,
      is_weighted: bool = False,
  ) -> Perf:
      batch_inputs = sum(
          [
              x * y * z
              for x, y, z in zip(
                  input_lengths, num_poolings, batch_sizes
              )
          ]
      )
      batch_outputs = (
          sum([x * y for x, y in zip(num_poolings, batch_sizes)])
          if is_pooled
          else batch_inputs
      )
  
      input_read_size = math.ceil(batch_inputs * input_data_type_size)
      if is_weighted:
          input_read_size *= 2
  
      embedding_lookup_size = (
          batch_inputs * emb_dim * table_data_type_size
      )
  
      output_write_size = batch_outputs * emb_dim * table_data_type_size
      table_size = grad_num_elem * table_data_type_size
  
      fwd_compute = (
          input_read_size + embedding_lookup_size + output_write_size
      ) / device_bw
  
      num_nodes = min(world_size / local_world_size, 2)
  
      # all-reduce data transfer: https://images.nvidia.com/events/sc15/pdfs/NCCL-Woolley.pdf
      all_reduce = (
          table_size
          * (2 * num_nodes - 1)
          / num_nodes
          / (inter_host_bw * local_world_size)  # 1 NIC per GPU
      )
      # inter host communication constraint
      if world_size > 2 * local_world_size:
          all_reduce *= 2
  
      # SGD + Fill + BUnary
      optimizer_kernels = (
          table_size * DP_ELEMENTWISE_KERNELS_PERF_FACTOR / device_bw
      )
  
      bwd_compute = fwd_compute * bwd_compute_multiplier
  
      bwd_grad_indice_weights_kernel = (
          fwd_compute * WEIGHTED_KERNEL_MULTIPLIER if is_weighted else 0
      )
  
      # TODO(T170641643): we don't model prefetch_compute for data parallel yet, see
      # comment in perf_func_emb_wall_time() regarding expected_cache_fetches calculation.
      return Perf(
          fwd_compute=fwd_compute,
          fwd_comms=0,
          bwd_compute=bwd_compute + bwd_grad_indice_weights_kernel,
          bwd_comms=all_reduce + optimizer_kernels,
      )
  
imports: import logging
  import math
  from typing import cast, Dict, List, Optional, Tuple, Type
  import torch
  import torchrec.optim
  from torch import nn
  from torchrec.distributed.embedding_types import EmbeddingComputeKernel
  from torchrec.distributed.planner.constants import BATCHED_COPY_PERF_FACTOR, BIGINT_DTYPE, DP_ELEMENTWISE_KERNELS_PERF_FACTOR, FULL_BLOCK_EMB_DIM, HALF_BLOCK_PENALTY, kernel_bw_lookup, QUARTER_BLOCK_PENALTY, UVM_CACHING_RATIO, WEIGHTED_KERNEL_MULTIPLIER
  from torchrec.distributed.planner.types import ParameterConstraints, Perf, PlannerError, ShardEstimator, ShardingOption, Storage, Topology
  from torchrec.distributed.planner.utils import prod, sharder_name
  from torchrec.distributed.types import CacheStatistics, CommOp, ModuleSharder, ShardingType
  from torchrec.modules.embedding_modules import EmbeddingBagCollectionInterface
label_desc: Use image node_id 6 for calling a global function with example usage: _get_dp_sharding_perf(batch_sizes, world_size, local_world_size, input_lengths, grad_num_elem, emb_dim, input_data_type_size, table_data_type_size, num_poolings, device_bw, inter_host_bw, bwd_compute_multiplier, is_pooled, is_weighted) and returns: Perf