code: def _run_test_dist(
      cls,
      rank: int,
      world_size: int,
      _input: torch.Tensor,
      output: torch.Tensor,
      input_splits: List[int],
      output_splits: List[int],
      lengths_after_sdd_a2a: torch.Tensor,
      features_per_rank: List[int],
      batch_size_per_rank: List[int],
      qcomms_config: Optional[QCommsConfig] = None,
  ) -> None:
      dist.init_process_group(
          rank=rank, world_size=world_size, backend="nccl"
      )
      pg = dist.group.WORLD
      device = torch.device(f"cuda:{rank}")
      _input = _input.to(device=device)
      output = output.to(device=device)
      lengths_after_sdd_a2a = lengths_after_sdd_a2a.to(device=device)
  
      a2a = SequenceEmbeddingsAllToAll(
          # pyre-fixme[6]: For 1st param expected `ProcessGroup` but got
          #  `Optional[_distributed_c10d.ProcessGroup]`.
          pg=pg,
          features_per_rank=features_per_rank,
          device=device,
      )
      _input.requires_grad = True
  
      sparse_features_recat = (
          _get_recat(
              local_split=features_per_rank[rank],
              num_splits=world_size,
              device=device,
              stagger=1,
              batch_size_per_rank=batch_size_per_rank,
          )
          if len(set(batch_size_per_rank)) > 1
          else None
      )
  
      res = a2a(
          local_embs=_input,
          lengths=lengths_after_sdd_a2a,
          input_splits=input_splits,
          output_splits=output_splits,
          batch_size_per_rank=batch_size_per_rank,
          sparse_features_recat=sparse_features_recat,
      ).wait()
  
      atol, rtol = None, None
      if qcomms_config is not None:
          atol, rtol = 0.01, 0.01
          if (
              qcomms_config.forward_precision == CommType.FP8
              or qcomms_config.backward_precision == CommType.FP8
          ):
              atol, rtol = 0.05, 0.05
      torch.testing.assert_close(res, output, rtol=rtol, atol=atol)
      res.backward(res)
      grad = _input.grad
      torch.testing.assert_close(
          _input.cpu().detach(),
          # pyre-fixme[16]: Optional type has no attribute `cpu`.
          grad.cpu().detach() * world_size,
      )
  
imports: import itertools
  import random
  import unittest
  from typing import Dict, Generator, Iterable, List, Optional, Tuple, TypeVar, Union
  import hypothesis.strategies
  import torch
  import torch.distributed
  from hypothesis import given, settings
  from torchrec.distributed.dist_data import _get_recat, KJTAllToAll, KJTAllToAllSplitsAwaitable, PooledEmbeddingsAllGather, PooledEmbeddingsAllToAll, PooledEmbeddingsReduceScatter, SequenceEmbeddingsAllToAll, VariableBatchPooledEmbeddingsAllToAll
  from torchrec.distributed.fbgemm_qcomm_codec import CommType, get_qcomm_codecs, QCommsConfig
  from torchrec.distributed.test_utils.multi_process import MultiProcessTestBase
  from torchrec.sparse.jagged_tensor import KeyedJaggedTensor
label_desc: Use image node_id 1 for calling the SeqEmbeddingsAllToAllTest obj's underlying member method code with example usage: obj._run_test_dist(cls, rank, world_size, _input, output, input_splits, output_splits, lengths_after_sdd_a2a, features_per_rank, batch_size_per_rank, qcomms_config) without return types