code: def run_partition_server(
      config: ConfigSchema,
      rank: Rank = SINGLE_TRAINER,
      subprocess_init: Optional[Callable[[], None]] = None,
  ) -> None:
      tag_logs_with_process_name(f"PartS-{rank}")
      if config.num_partition_servers <= 0:
          raise RuntimeError(
              "Config doesn't require explicit partition servers"
          )
      if not 0 <= rank < config.num_partition_servers:
          raise RuntimeError("Invalid rank for partition server")
      if not td.is_available():
          raise RuntimeError(
              "The installed PyTorch version doesn't provide "
              "distributed training capabilities."
          )
      ranks = ProcessRanks.from_num_invocations(
          config.num_machines, config.num_partition_servers
      )
  
      if config.num_partition_servers > config.num_machines:
          num_ps_groups = config.num_partition_servers * (
              config.num_groups_per_sharded_partition_server + 1
          )
      else:
          num_ps_groups = config.num_groups_for_partition_server
      groups: List[List[int]] = [ranks.trainers]  # barrier group
      groups += [
          ranks.trainers + ranks.partition_servers
      ] * num_ps_groups  # ps groups
      group_idxs_for_partition_servers = range(1, len(groups))
  
      if subprocess_init is not None:
          subprocess_init()
      groups = init_process_group(
          rank=ranks.partition_servers[rank],
          world_size=ranks.world_size,
          init_method=config.distributed_init_method,
          groups=groups,
      )
      if config.num_partition_servers > config.num_machines:
          ps = ShardedParameterServer(
              num_clients=len(ranks.trainers),
              num_data_pgs=config.num_groups_per_sharded_partition_server,
              group_idxs=group_idxs_for_partition_servers,
              log_stats=True,
          )
      else:
          ps = ParameterServer(
              num_clients=len(ranks.trainers),
              group_idxs=group_idxs_for_partition_servers,
              log_stats=True,
          )
      ps.start(groups)
      logger.info("ps.start done")
  
file_path: PyTorch-BigGraph/torchbiggraph/partitionserver.py
imports: import argparse
  import logging
  from typing import Callable, List, Optional
  import torch.distributed
  from torchbiggraph.config import add_to_sys_path, ConfigFileLoader, ConfigSchema
  from torchbiggraph.distributed import init_process_group, ProcessRanks
  from torchbiggraph.parameter_sharing import ParameterServer, ShardedParameterServer
  from torchbiggraph.types import Rank, SINGLE_TRAINER
  from torchbiggraph.util import set_logging_verbosity, setup_logging, SubprocessInitializer, tag_logs_with_process_name
label_desc: Use image node_id 1 for calling a global function with example usage: run_partition_server(config, rank, subprocess_init) without return types