class_docstr: Embedding Wall Time Perf Estimator
code: def estimate(
      self,
      sharding_options: List[ShardingOption],
      sharder_map: Optional[Dict[str, ModuleSharder[nn.Module]]] = None,
  ) -> None:
      if not sharder_map:
          assert (
              not sharding_options
          ), "sharder_map not provided for sharding_options"
          return
  
      for sharding_option in sharding_options:
          sharder_key = sharder_name(type(sharding_option.module[1]))
          sharder = sharder_map[sharder_key]
  
          caching_ratio = (
              sharding_option.cache_params.load_factor
              if sharding_option.cache_params
              else None
          )
          # TODO: remove after deprecating fused_params in sharder
          if caching_ratio is None:
              caching_ratio = (
                  sharder.fused_params.get(
                      "cache_load_factor"
                  )  # pyre-ignore[16]
                  if hasattr(sharder, "fused_params")
                  and sharder.fused_params
                  else None
              )
  
          num_poolings = (
              cast(
                  List[float],
                  self._constraints[sharding_option.name].num_poolings,
              )
              if self._constraints
              and self._constraints.get(sharding_option.name)
              and self._constraints[sharding_option.name].num_poolings
              else [1.0] * sharding_option.num_inputs
          )
          batch_sizes = (
              cast(
                  List[int],
                  self._constraints[sharding_option.name].batch_sizes,
              )
              if self._constraints
              and self._constraints.get(sharding_option.name)
              and self._constraints[sharding_option.name].batch_sizes
              else [sharding_option.batch_size]
              * sharding_option.num_inputs
          )
  
          assert (
              len(sharding_option.input_lengths)
              == len(num_poolings)
              == len(batch_sizes)
          ), "Provided `pooling_factors`, `num_poolings`, and `batch_sizes` constraints must match."
  
          module = sharding_option.module[1]
  
          # TODO remove this hack once feature processor is disaggregated
          has_feature_processor = False
          if (
              hasattr(module, "_feature_processor")
              and hasattr(
                  module._feature_processor, "feature_processor_modules"
              )
              and isinstance(
                  module._feature_processor.feature_processor_modules,
                  nn.ModuleDict,
              )
              and sharding_option.name
              in module._feature_processor.feature_processor_modules.keys()
          ):
              has_feature_processor = True
              logger.info(
                  f"Table {sharding_option.name} has feature processor."
              )
  
          if isinstance(module, EmbeddingBagCollectionInterface):
              is_weighted = module.is_weighted()
          elif (
              self._constraints
              and self._constraints.get(sharding_option.name)
              and self._constraints[sharding_option.name].is_weighted
          ):
              is_weighted = self._constraints[
                  sharding_option.name
              ].is_weighted
          else:
              is_weighted = False
  
          # TODO remove this once migrate away from PEA
          is_weighted = is_weighted or has_feature_processor
          sharding_option.is_weighted = is_weighted
  
          table_data_type_size = sharding_option.tensor.element_size()
          (
              fwd_a2a_comm_data_type_size,
              bwd_a2a_comm_data_type_size,
              fwd_sr_comm_data_type_size,
              bwd_sr_comm_data_type_size,
          ) = _extract_comm_data_type_size(sharder, sharding_option)
  
          prefetch_pipeline = (
              sharding_option.cache_params.prefetch_pipeline
              if sharding_option.cache_params
              else None
          )
          # TODO: remove after deprecating fused_params in sharder
          if not prefetch_pipeline:
              prefetch_pipeline = (
                  sharder.fused_params.get("prefetch_pipeline", False)
                  if hasattr(sharder, "fused_params")
                  and sharder.fused_params
                  else False
              )
  
          expected_cache_fetches = 0
          if (
              caching_ratio is not None
              and sharding_option.cache_params is not None
              and sharding_option.cache_params.stats is not None
              and sharding_option.compute_kernel
              == EmbeddingComputeKernel.FUSED_UVM_CACHING.value
          ):
              _stats = sharding_option.cache_params.stats
              expected_cache_fetches = (
                  _stats.expected_miss_rate(caching_ratio)
                  * _stats.expected_lookups
              )
              # Note, the above is not correct for data-parallel. stats.expected_lookups is
              # calculated by estimating the cardinality of a global batch size worth of data.
              # But for data-parallel, we need the calculate the cardinality of the local
              # input batch.  For now, we don't use cache stats with data parallel.
  
          shard_perfs = perf_func_emb_wall_time(
              shard_sizes=[
                  shard.size for shard in sharding_option.shards
              ],
              compute_kernel=sharding_option.compute_kernel,
              compute_device=self._topology.compute_device,
              sharding_type=sharding_option.sharding_type,
              batch_sizes=batch_sizes,
              world_size=self._topology.world_size,
              local_world_size=self._topology.local_world_size,
              input_lengths=sharding_option.input_lengths,
              input_data_type_size=BIGINT_DTYPE,
              table_data_type_size=table_data_type_size,
              fwd_a2a_comm_data_type_size=fwd_a2a_comm_data_type_size,
              bwd_a2a_comm_data_type_size=bwd_a2a_comm_data_type_size,
              fwd_sr_comm_data_type_size=fwd_sr_comm_data_type_size,
              bwd_sr_comm_data_type_size=bwd_sr_comm_data_type_size,
              num_poolings=num_poolings,
              hbm_mem_bw=self._topology.hbm_mem_bw,
              ddr_mem_bw=self._topology.ddr_mem_bw,
              intra_host_bw=self._topology.intra_host_bw,
              inter_host_bw=self._topology.inter_host_bw,
              bwd_compute_multiplier=self._topology.bwd_compute_multiplier,
              is_pooled=sharding_option.is_pooled,
              is_weighted=is_weighted,
              is_inference=self._is_inference,
              caching_ratio=caching_ratio,
              prefetch_pipeline=prefetch_pipeline,
              expected_cache_fetches=expected_cache_fetches,
          )
  
          for shard, perf in zip(sharding_option.shards, shard_perfs):
              shard.perf = perf
  
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 2 for calling the EmbeddingPerfEstimator obj's underlying member method code with example usage: obj.estimate(sharding_options, sharder_map) without return types