Skip to content

Support torchrun-style InfiniTrain multi-process launch - #184

Open
chen2021673 wants to merge 6 commits into
masterfrom
8_proc
Open

Support torchrun-style InfiniTrain multi-process launch#184
chen2021673 wants to merge 6 commits into
masterfrom
8_proc

Conversation

@chen2021673

@chen2021673 chen2021673 commented Jul 3, 2026

Copy link
Copy Markdown
Contributor

Summary

InfiniTrain’s existing parallel execution model primarily launches multiple training threads within a single process. This PR adds a torchrun-style multi-process launcher, allowing each local process to bind to its own GPU while preserving the existing intra-process multithreading mode. It also fixes DataLoader and NCCL unique ID file conflicts in multi-process environments.

  • Add support for launching training processes through infini_run using nproc_per_node.
  • Ensure that the NCCL unique ID is generated only by the communication group’s root rank, with per-run file isolation, atomic publication, and cleanup.
  • Use global batches as the unit for distributed DataLoader partitioning and retrieval, aligning with Megatron’s behavior and fixing out-of-range batch access.
  • Add an 8-process integration test.

Changes

  • Update infini_run to:

    • support -- as the launcher/training-args separator
    • launch nproc_per_node child processes
    • inject both InfiniTrain and torchrun-compatible rank env vars
    • propagate child process failures via exit code
  • Update parallel runtime to:

    • read torchrun-compatible env vars as fallback
    • validate process topology and rank bounds
    • map local process/thread rank to CUDA device index
  • Update GPT-2/Llama3 examples and parallel helpers to use local-device mapping.

  • Update scripts/run_models_and_profile.bash to:

    • always launch model commands through infini_run
    • treat nproc_per_node as launcher-only config
    • keep nthread_per_process as the per-process thread count
  • Update scripts/test_config.json to use multi-process configs:

    • 8-thread cases become nproc_per_node=8, nthread_per_process=1
    • original 4-rank VPP cases become nproc_per_node=4, nthread_per_process=1
  • Add documentation describing behavior, compatibility, and example usage.

Compatibility

Existing direct runs remain supported:

./llama3 ... --nthread_per_process 8

The launcher can also preserve the old single-process multi-thread behavior:

./infini_run --nproc_per_node=1 ./llama3 ... --nthread_per_process 8

The recommended single-node 8-GPU multi-process usage is:

./infini_run --nproc_per_node=8 ./llama3 ... --nthread_per_process 1

Test

image image

Comment thread example/gpt2/main.cc
Comment thread infini_train/include/core/ccl/ccl.h Outdated
virtual void GetAsyncError(const CclComm *comm, CclStatus *async_error) const;

virtual void GetUniqueId(CclUniqueId **unique_id) const;
virtual void CreateUniqueId(CclUniqueId **unique_id, bool generate_id) const;

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

这个接口的名字和参数都不建议改,因为对标的是 nccl 接口 GetUniqueId() ,后续国产平台的应该也都是类似签名。

下面的 nccl_impl.h 的继承实现也得改回来。

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

done

Comment thread tools/infini_run/infini_run.cc Outdated
SetEnvInt("WORLD_SIZE", proc_world_size);
SetEnvInt("GROUP_RANK", FLAGS_node_rank);
SetEnvInt("ROLE_RANK", global_proc_rank);
SetEnvInt("ROLE_WORLD_SIZE", proc_world_size);

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

这几个好像没用到?可以先删掉

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

done

}
} else if (exit_code == 0) {
exit_code = 1;
}

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

这块退出,好像也没有做某个子进程异常退出的时候清理其他进程的逻辑?如果 exit code 非 0 的话感觉正常情况应该要把所有其他正在运行的子进程都清理完毕再返回

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

修改逻辑为:跟踪所有运行中的子进程;任一子进程异常退出或 fork 失败时,向其余进程发送 SIGTERM,继续回收全部子进程后返回首个失败码。

Comment thread tools/infini_run/infini_run.cc Outdated
int proc_world_size = FLAGS_nnodes * FLAGS_nproc_per_node;
std::string master_addr = FLAGS_rdzv_endpoint.substr(0, FLAGS_rdzv_endpoint.find(':'));
std::string master_port = FLAGS_rdzv_endpoint.substr(FLAGS_rdzv_endpoint.find(':') + 1);
const std::string run_id = FLAGS_nnodes == 1 ? GenerateLocalRunId() : "";

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

这块多机还是会使用原先的默认命名,没达到效果,可能得看下怎么改。

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

修改:多机必须通过 --rdzv_id 指定;单机未指定时仍自动生成。参考https://docs.pytorch.org/docs/2.13/elastic/run.html

Add a dedicated 8_proc test group containing the 8-process variants of
the original basic multi-GPU cases.
Track DataLoader progress by global batches so distributed ranks slice data
consistently and can resume/cycle from saved consumption counts.

Also scope CCL unique ID files per run, generate NCCL IDs only on the main
rank, clean up run-local rendezvous files, and add DataLoader coverage.
- derive parallel state from the global world size
- clarify global rank and per-node process semantics
- add multi-node rank regression coverage
- restore the NCCL-compatible GetUniqueId interface
- add torchrun-style --rdzv_id support
- use the shared ID to isolate CCL unique-ID files
- preserve automatic run ID generation for single-node runs
- document rdzv_id in the multi-node example
Comment thread README.md
--nnodes=2 \
--nproc_per_node=1 \
--node_rank=[rank_id] \
-- ./llama3 \

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

需要确认下目前的使用行为,README 里删除了 -- 作为启动/训练参数的分隔符?但我看 pr description 里写的是 "support -- as the launcher/training-args separator",原始版本也是通过 -- 作为分隔符的。

顺便指出 pr description 的另一处小问题:
Image
这里似乎需要更新一下,目前应该是多进程/多线程测例同时保留了,vpp case 的说明似乎也和实际代码不一致。

<< "GLOBAL_PROC_RANK/RANK must be less than PROC_WORLD_SIZE/WORLD_SIZE";
CHECK_GE(local_proc_rank_, 0) << "LOCAL_PROC_RANK/LOCAL_RANK must be non-negative";
CHECK_LT(local_proc_rank_, nproc_per_node_)
<< "LOCAL_PROC_RANK/LOCAL_RANK must be less than NPROC_PER_NODE/LOCAL_WORLD_SIZE";

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

https://docs.pytorch.org/docs/2.13/elastic/run.html#environment-variables
修改后更加贴近 torchrun 的环境变量命名,统一保留新名称即可,不需要兼容旧环境变量名

for (int index = 0; index < global::GetNthreadPerProc(); ++index) { devices_.emplace_back(device_type, index); }
for (int thread_rank = 0; thread_rank < global::GetNthreadPerProc(); ++thread_rank) {
const int device_index
= device_type == Device::DeviceType::kCUDA ? global::GetLocalDeviceIndex(thread_rank) : thread_rank;

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

不需要特判 cuda 类型,其他硬件遵循 cuda 分配 device index 的规则。

inline int GetNthreadPerProc() { return GlobalEnv::Instance().nthread_per_process(); }
inline int GetGlobalProcRank() { return GlobalEnv::Instance().global_proc_rank(); }
inline int GetLocalProcRank() { return GlobalEnv::Instance().local_proc_rank(); }
inline int GetLocalDeviceIndex(int thread_rank = 0) { return GetLocalProcRank() * GetNthreadPerProc() + thread_rank; }

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

device index 本身就特指节点内的设备编号,是不是没必要强调 local 了;
以及这个函数感觉不适合有默认值。

if (IsCPU()) {
return {nn::parallel::global::GetGlobalProcRank(), 0, nn::parallel::global::GetNprocPerNode(),
nn::parallel::global::GetNthreadPerProc()};
}

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

cpu 的情况在 Device 的构造函数里已经处理过了,这里不需要特判。

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

dataloader 相关的改动应该与多进程启动无关,建议拆出独立 pr 进行 review 和合入流程。

namespace {
std::string UniqueIdFileName(const std::string &name, bool tmp = false) {
return "cclUniqueId_" + name + (tmp ? ".tmp" : ".bin");
std::string UniqueIdPath(const std::string &pg_name) {

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

这里生成的确实是 file name 而不是 path 吧,应该不需要修改函数名,下面 tmp 函数同理。

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

按照之前讨论的结论,现有测例应保持不变,仅新增 8_proc 测例采用 infini_run 启动多进程分布式任务,但目前脚本将所有测例都改成统一使用 infini_run 启动了,麻烦修改一下。


std::string train_program = argv[1];
std::string train_program = argv[train_program_index];
CHECK_NE(train_program, "--") << "Explicit '--' separator is not supported; pass the training program directly "

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

使用 -- 分隔符调用 infini_run 时,前面 FindTrainProgramIndex 返回 -- 的位置,但这里又明确拒绝了 --,导致使用 -- 分隔训练参数的使用方式会直接报错,建议同时支持使用/不使用 -- 分隔的调用方式。


CHECK_GE(argc, 2) << "No training prgram specified!";
CHECK_GT(FLAGS_nnodes, 0) << "nnodes must be positive";
CHECK_GT(FLAGS_nproc_per_node, 0) << "nproc_per_node must be positive";

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

再检查下

CHECK_GE(FLAGS_node_rank, 0);
CHECK_LT(FLAGS_node_rank, FLAGS_nnodes);

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants