在DI-engine中使用DDP分布式训练¶

分布式数据并行（DDP）是加速强化学习训练的有效方法。本文档以pong_dqn_ddp_config.py为例，提供了在DI-engine中配置和使用DDP训练的详细说明。

启动DDP训练¶

要启动DDP训练，请使用PyTorch的distributed.launch模块。运行以下命令：

python -m torch.distributed.launch --nproc_per_node=2 --master_port=29501 ./dizoo/atari/config/serial/pong/pong_dqn_ddp_config.py

其中：

--nproc_per_node=2: 指定使用2个GPU进行训练。
--master_port=29501: 指定主进程的端口号。
最后一个参数是配置文件的路径。

DDP训练的配置¶

与单GPU训练（例如，pong_dqn_config.py）相比，启用多GPU DDP训练的配置文件pong_dqn_ddp_config.py引入了两个关键变化：

在策略配置中启用多GPU支持:
```
policy = dict(
    multi_gpu=True,  # Enable multi-GPU training mode
    cuda=True,       # Use CUDA acceleration
    ...
)
```
- 初始化多GPU训练的核心代码位于base_policy.py: base_policy.py#L167
- 梯度同步发生在 policy._forward_learn(): dqn.py#L281

使用 ``DDPContext`` 来管理分布式训练过程:

if __name__ == '__main__':
    from ding.utils import DDPContext
    from ding.entry import serial_pipeline
    with DDPContext():
        serial_pipeline((main_config, create_config), seed=0, max_env_step=int(3e6))

DDPContext 初始化分布式训练环境，并在训练完成后释放分布式资源。

DI-engine中的DDP实现¶

DI-engine中的DDP实现包括以下关键组件：

收集器中数据收集的分布式处理:

在SampleSerialCollector中，每个进程独立收集数据样本。

收集后，统计数据通过allreduce在进程间同步：

if self._world_size > 1:
    collected_sample = allreduce_data(collected_sample, 'sum')
    collected_step = allreduce_data(collected_step, 'sum')
    collected_episode = allreduce_data(collected_episode, 'sum')
    collected_duration = allreduce_data(collected_duration, 'sum')

参见 sample_serial_collector.py#L355

评估器的分布式处理:

评估逻辑仅在排名为0的进程上运行：
```
if get_rank() == 0:
    # Perform evaluation logic
    ...
```
- 参见 interaction_serial_evaluator.py#L207

评估后，结果将广播到其他进程：

if get_world_size() > 1:
    objects = [stop_flag, episode_info]
    broadcast_object_list(objects, src=0)
    stop_flag, episode_info = objects

参见 interaction_serial_evaluator.py#L315

分布式日志记录:

日志记录器仅在排名为0的进程上初始化：

tb_logger = SummaryWriter(os.path.join('./{}/log/'.format(cfg.exp_name), 'serial')) if get_rank() == 0 else None

参见 serial_entry.py#L72

日志记录仅限于排名为0的进程：

if self._rank == 0:
    if tb_logger is not None:
        self._logger, _ = build_logger(
            path='./{}/log/{}'.format(self._exp_name, self._instance_name),
            name=self._instance_name,
            need_tb=False
        )
        self._tb_logger = tb_logger
    else:
        self._logger, self._tb_logger = build_logger(
            path='./{}/log/{}'.format(self._exp_name, self._instance_name),
            name=self._instance_name
        )
else:
    self._logger, _ = build_logger(
        path='./{}/log/{}'.format(self._exp_name, self._instance_name),
        name=self._instance_name,
        need_tb=False
    )
    self._tb_logger = None

参见 sample_serial_collector.py#L59

打印日志也仅限于排名为0的进程：
```
if self._rank != 0:
    return
```
- 参见 sample_serial_collector.py#L388

摘要¶

在DI-engine中，DDP分布式训练充分利用了多个GPU的计算能力，通过分布式数据收集、评估和日志记录来加速训练过程。DDP的核心逻辑依赖于PyTorch的分布式框架，而DDPContext提供了对分布式环境的统一管理，简化了开发者的配置和使用过程。

有关实现的更多详细信息，请参考以下代码引用：