class_docstr: :param model: 训练使用的模型。
  
      * 如果不想自己初始化分布式环境，类型应为 :class:`paddle.nn.Layer`；
      * 如果已经在外面初始化了分布式环境，类型应为 :class:`paddle.DataParallel`；
  
  :param parallel_device: 多卡训练时使用的设备，必须是一个列表。
      当使用 ``python -m paddle.distributed.launch`` 启动时，该参数无效。
  :param is_pull_by_paddle_run: 标记当前进程是否为通过 ``python -m paddle.distributed.launch`` 启动的。
      这个参数仅在 :class:`~fastNLP.core.Trainer` 中初始化 driver 时使用
  :param fp16: 是否开启混合精度训练
  :param paddle_kwargs:
      * *fleet_kwargs* -- 用于在使用 ``PaddleFleetDriver`` 时指定 ``DataParallel`` 和 ``fleet`` 初始化时的参数，包括：
          
          * *is_collective* -- 是否使用 paddle 集群式的分布式训练方法，目前仅支持为 ``True`` 的情况。
          * *role_maker* -- 初始化 ``fleet`` 分布式训练 API 时使用的 ``RoleMaker``。
          * 其它用于初始化 ``DataParallel`` 的参数。
      * *gradscaler_kwargs* -- 用于 ``fp16=True`` 时，提供给 :class:`paddle.amp.GradScaler` 的参数
  
  :kwargs:
      * *wo_auto_param_call* (``bool``) -- 是否关闭在训练时调用我们的 ``auto_param_call`` 函数来自动匹配 batch 和前向函数的参数的行为
  
      .. note::
  
          关于该参数的详细说明，请参见 :class:`~fastNLP.core.controllers.Trainer` 中的描述；函数 ``auto_param_call`` 详见 :func:`fastNLP.core.utils.auto_param_call`。
docstr: 初始化分布式训练的环境。
  
  1. 如果是通过 ``paddle.distributed.launch`` 方法启动的，则根据已经设置好的环境获取分布式的属性。
  2. 否则启动子进程。
code: def setup(self):
      """
      初始化分布式训练的环境。
  
      1. 如果是通过 ``paddle.distributed.launch`` 方法启动的，则根据已经设置好的环境获取分布式的属性。
      2. 否则启动子进程。
      """
      if self._has_setup:
          return
      self._has_setup = True
      # 如果用户需要使用多机模式，那么一定进入到这里；
      if self.is_pull_by_paddle_run:
          if self.outside_fleet:
              # 已经初始化了多机环境
              self._set_from_fleet_environment()
          else:
              # 用户没有初始化多机环境
              # TODO 绕一下
              # dist.get_world_size() 只能在初始化之后进行调用；
              self.world_size = int(
                  os.environ.get("PADDLE_TRAINERS_NUM")
              )
              self.global_rank = int(
                  os.environ.get("PADDLE_TRAINER_ID")
              )
              logger.info(
                  f"World size: {self.world_size}, Global rank: {self.global_rank}"
              )
              if not parallel_helper._is_parallel_ctx_initialized():
                  fleet.init(
                      self.role_maker, self.is_collective, self.strategy
                  )
  
              os.environ["fastnlp_paddle_launch_not_fleet"] = "yes"
  
      else:
          # 在用户只使用了一个分布式 trainer 的情况下
          # 此时 parallel_helper._is_parallel_ctx_initialized() 一定为 False
          # parallel_device 是 list，
          if not parallel_helper._is_parallel_ctx_initialized():
              # 拉起子进程并设置相应的属性
              self._init_fleet_and_set()
          # 用户在这个 trainer 前面又初始化了一个 trainer，并且使用的是 PaddleFleetDriver；
          else:
              # 已经设置过一次，保证参数必须是一样的
              pre_gpus = os.environ[FASTNLP_DISTRIBUTED_CHECK]
              pre_gpus = [int(x) for x in pre_gpus.split(",")]
              cur_gpus = [
                  get_paddle_device_id(g) for g in self.parallel_device
              ]
              if sorted(pre_gpus) != sorted(self.parallel_device):
                  raise RuntimeError(
                      "Notice you are using `PaddleFleetDriver` after one instantiated `PaddleFleetDriver`, it is not"
                      "allowed that your second `PaddleFleetDriver` has a new setting of parameters `parallel_device`."
                  )
              self.world_size = paddledist.get_world_size()
              self.global_rank = paddledist.get_rank()
  
      if not self.outside_fleet:
          # self.model.to(self.model_device)
          self.configure_fleet()
  
      self.barrier()
  
      # 初始化 self._pids，从而使得每一个进程都能接受到 rank0 的 send 操作；
      # TODO 不用.to会怎么样？
      self._pids = []
      paddledist.all_gather(
          self._pids, paddle.to_tensor(os.getpid(), dtype="int32")
      )
      # TODO LOCAL_WORLD_SIZE
      local_world_size = (
          int(os.environ.get("LOCAL_WORLD_SIZE"))
          if "LOCAL_WORLD_SIZE" in os.environ
          else None
      )
      if local_world_size is None:
          local_world_size = paddle.to_tensor(
              self.local_rank, dtype="int32"
          )
          paddledist.all_reduce(
              local_world_size, op=paddledist.ReduceOp.MAX
          )
          local_world_size = local_world_size.item() + 1
  
      node_rank = self.global_rank // local_world_size
      self._pids = self._pids[
          node_rank
          * local_world_size : (node_rank + 1)
          * local_world_size
      ]
      self._pids = self.tensor_to_numeric(self._pids)
  
imports: import os
  from typing import List, Union, Optional, Dict, Tuple, Callable
  from .paddle_driver import PaddleDriver
  from .fleet_launcher import FleetLauncher
  from .utils import _FleetWrappingModel, replace_sampler, replace_batch_sampler, _check_dataloader_args_for_distributed
  from .dist_utils import fastnlp_paddle_all_gather, fastnlp_paddle_broadcast_object
  from fastNLP.envs.imports import _NEED_IMPORT_PADDLE
  from fastNLP.core.utils import auto_param_call, check_user_specific_params, is_in_paddle_dist, get_paddle_device_id
  from fastNLP.core.utils.paddle_utils import _convert_data_device
  from fastNLP.envs.distributed import rank_zero_rm
  from fastNLP.core.samplers import ReproduceBatchSampler, ReproducibleSampler, ReproducibleBatchSampler, RandomSampler, UnrepeatedSampler, UnrepeatedSequentialSampler, re_instantiate_sampler, conversion_between_reproducible_and_unrepeated_sampler
  from fastNLP.envs.env import FASTNLP_DISTRIBUTED_CHECK, FASTNLP_GLOBAL_SEED, FASTNLP_NO_SYNC, USER_CUDA_VISIBLE_DEVICES
  from fastNLP.core.log import logger
label_desc: Use image node_id 2 for calling the PaddleFleetDriver obj's underlying member method code with example usage: obj.setup() without return types