class_docstr: Base class for table wise row wise sharding.
class_docstr_tok: ["Base","class","for","table","wise","row","wise","sharding","."]
code: def __init__(
      self,
      sharding_infos: List[EmbeddingShardingInfo],
      env: ShardingEnv,
      device: Optional[torch.device] = None,
      need_pos: bool = False,
      qcomm_codecs_registry: Optional[
          Dict[str, QuantizedCommCodecs]
      ] = None,
  ) -> None:
      super().__init__(qcomm_codecs_registry=qcomm_codecs_registry)
      self._env = env
      self._pg: Optional[dist.ProcessGroup] = self._env.process_group
      self._world_size: int = self._env.world_size
      self._rank: int = self._env.rank
      self._device = device
      self._need_pos = need_pos
      intra_pg, cross_pg = intra_and_cross_node_pg(
          device, backend=dist.get_backend(self._pg)
      )
      self._intra_pg: Optional[dist.ProcessGroup] = intra_pg
      self._cross_pg: Optional[dist.ProcessGroup] = cross_pg
      self._local_size: int = (
          intra_pg.size()
          if intra_pg
          else get_local_size(self._world_size)
      )
  
      sharded_tables_per_rank = self._shard(sharding_infos)
      self._grouped_embedding_configs_per_rank: List[
          List[GroupedEmbeddingConfig]
      ] = []
      self._grouped_embedding_configs_per_node: List[
          List[GroupedEmbeddingConfig]
      ] = []
      self._grouped_embedding_configs_per_rank = group_tables(
          sharded_tables_per_rank
      )
      self._grouped_embedding_configs_per_node = [
          self._grouped_embedding_configs_per_rank[rank]
          for rank in range(self._world_size)
          if rank % self._local_size == 0
      ]
      self._has_feature_processor: bool = False
      for group_config in self._grouped_embedding_configs_per_rank[
          self._rank // self._local_size
      ]:
          if group_config.has_feature_processor:
              self._has_feature_processor = True
  
imports: import itertools
  import math
  from typing import Any, cast, Dict, List, Optional, Tuple, TypeVar, Union
  import torch
  import torch.distributed
  from torchrec.distributed.comm import get_local_size, intra_and_cross_node_pg
  from torchrec.distributed.dist_data import KJTAllToAll, PooledEmbeddingsAllToAll, PooledEmbeddingsReduceScatter, VariableBatchPooledEmbeddingsAllToAll, VariableBatchPooledEmbeddingsReduceScatter
  from torchrec.distributed.embedding_lookup import GroupedPooledEmbeddingsLookup
  from torchrec.distributed.embedding_sharding import BaseEmbeddingDist, BaseEmbeddingLookup, BaseSparseFeaturesDist, bucketize_kjt_before_all2all, EmbeddingSharding, EmbeddingShardingContext, EmbeddingShardingInfo, group_tables
  from torchrec.distributed.embedding_types import BaseGroupedFeatureProcessor, EmbeddingComputeKernel, GroupedEmbeddingConfig, ShardedEmbeddingTable
  from torchrec.distributed.types import Awaitable, CommOp, QuantizedCommCodecs, ShardedTensorMetadata, ShardingEnv, ShardMetadata
  from torchrec.sparse.jagged_tensor import KeyedJaggedTensor
  from torchrec.streamable import Multistreamable
label_desc: Use image node_id 1 to create a new BaseTwRwEmbeddingSharding object from inherited base classes: EmbeddingSharding with example: obj = BaseTwRwEmbeddingSharding(sharding_infos, env, device, need_pos, qcomm_codecs_registry)