code: def _pipeline_parallel_post_init(
      cfg: DistributedTrainingConfig,
      num_pipeline_devices,
      num_pipelines_per_node,
  ):
      if not cfg.distributed_no_spawn:
          # When distributed_no_spawn is False, we expect distributed_rank and
          # distributed_world_size to be based on the total number of GPUs, so
          # we need to correct them to be based on the number of pipelines.
          assert cfg.distributed_world_size % num_pipeline_devices == 0
          cfg.distributed_world_size = (
              cfg.distributed_world_size // num_pipeline_devices
          )
          # In the case of 4-way MP on nodes with 8 GPUs, we want
          # distributed_rank to be the starting GPU index for each pipeline
          # i.e., 0, 2, ...
          gpus_per_node = torch.cuda.device_count()
          assert cfg.distributed_rank % gpus_per_node == 0
          assert cfg.distributed_rank % num_pipeline_devices == 0
  
          with open_dict(cfg):
              cfg.distributed_rank = (
                  cfg.distributed_rank // num_pipeline_devices
              )
              # launch one process per pipeline
              cfg.distributed_num_procs = num_pipelines_per_node
  
      # if we have 4-way MP on a node with 8 GPUs, we want device_ids to be 0
      # and 4, indicating the starting device IDs for each pipeline
      cfg.device_id *= num_pipeline_devices
  
      if cfg.device_id > 0:
          # if there's multiple pipelines on a node (e.g., 4-way MP on an 8
          # GPU node), we need to adjust pipeline_devices accordingly
          logger.debug(
              "setting CUDA device={} on rank {}".format(
                  cfg.device_id, cfg.distributed_rank
              )
          )
          torch.cuda.set_device(cfg.device_id)
          with open_dict(cfg):
              cfg.pipeline_devices = [
                  cfg.device_id + d for d in cfg.pipeline_devices
              ]
          logger.info(
              "setting pipeline_devices={} on rank {}".format(
                  cfg.pipeline_devices, cfg.distributed_rank
              )
          )
  
imports: import io
  import logging
  import os
  import pickle
  import random
  import socket
  import struct
  import subprocess
  import warnings
  from argparse import Namespace
  from collections import OrderedDict
  from dataclasses import dataclass
  from typing import Any, Dict, List, Mapping, Optional
  import torch
  import torch.distributed
  from fairseq.dataclass.configs import DistributedTrainingConfig, FairseqConfig
  from omegaconf import open_dict
label_desc: Use image node_id 7 for calling a global function with example usage: _pipeline_parallel_post_init(cfg, num_pipeline_devices, num_pipelines_per_node) without return types