code: def _create_expert_and_data_parallel(expert_parallel_size_):
      """
      Create expert and data parallel groups.
  
      Note: Caller of this function is responsible to check if the groups already exist.
  
      Example - E + D parallel
      world_size = 16
      expert_parallel_size = 2 # number of experts in same group
      expert_data_parallel_group = [0,2,4,6,8,10,12,14], [1,3,5,7,9,11,13,15] - all reduce is only on MoE params
      expert_parallel_group = [0, 1], [2,3], [4,5], [6,7], [8,9] - no all reduce, but all to all
      data_parallel_group = [0,1,...,15] - all reduce is only on non-MoE
      """
      assert dist.is_initialized()
  
      log_dist(
          f"Creating expert and data parallel groups with size {expert_parallel_size_}",
          ranks=[0],
      )
      world_size = dist.get_world_size()
      rank = dist.get_rank()
  
      _ensure_divisibility(world_size, expert_parallel_size_)
  
      group_name = f"ep_size_{expert_parallel_size_}"
  
      # Build the expert data parallel groups.
      global _EXPERT_DATA_PARALLEL_GROUP
  
      # Only create group if it does not already exist
      if group_name not in _EXPERT_DATA_PARALLEL_GROUP:
          for i in range(expert_parallel_size_):
              ranks = range(i, world_size, expert_parallel_size_)
              group = dist.new_group(ranks)
              log_dist(
                  f"Creating expert data parallel process group named {group_name} with ranks: {list(ranks)}",
                  [0],
              )
              if i == (rank % expert_parallel_size_):
                  _EXPERT_DATA_PARALLEL_GROUP[group_name] = group
  
      # Build the expert parallel groups.
      global _EXPERT_PARALLEL_GROUP
  
      # Only create group if it does not already exist
      if group_name not in _EXPERT_PARALLEL_GROUP:
          for i in range(world_size // expert_parallel_size_):
              ranks = range(
                  i * expert_parallel_size_,
                  (i + 1) * expert_parallel_size_,
              )
              group = dist.new_group(ranks)
              log_dist(
                  f"creating expert parallel process group named {group_name} with ranks: {list(ranks)}",
                  [0],
              )
              if i == (rank // expert_parallel_size_):
                  _EXPERT_PARALLEL_GROUP[group_name] = group
  
imports: from deepspeed import comm
  from deepspeed.utils import log_dist
  from deepspeed.utils.exceptions import DeprecatedException
label_desc: Use image node_id 4 for calling a global function with example usage: _create_expert_and_data_parallel(expert_parallel_size_) without return types