[Train] Add PyTorch local mode support for multi-process training with torchrun

[xinyuangui2]

This PR extends the Ray Train v2 local mode support (from #55487) to enable users to launch multiple local mode processes using torchrun for PyTorch distributed training. With this new feature, users can easily switch between torchrun and Ray Train without modifying their training code.

Note

Ray data on multiple processes is not supported. Might need to wait for #55114 or similar components.

Key Changes

Multi-Process Local Mode Support

• LocalTorchController: New controller that detects torchrun env variables and sets contexts accordingly
• Torchrun Integration: Users can now launch multiple local mode processes using torchrun command
• Environment Detection: Automatically detects torchrun environment variables and initializes distributed training

Usage Example

import os
import tempfile

import torch
from torch.nn import CrossEntropyLoss
from torch.optim import Adam
from torch.utils.data import DataLoader
from torchvision.models import resnet18
from torchvision.datasets import FashionMNIST
from torchvision.transforms import ToTensor, Normalize, Compose

import ray
from ray.train import Checkpoint, CheckpointConfig, RunConfig, ScalingConfig
from ray.train.torch import TorchTrainer
from ray.train.v2.api.config import FailureConfig
import ray.train.torch

def train_func():
    # Model, Loss, Optimizer
    model = resnet18(num_classes=10)
    model.conv1 = torch.nn.Conv2d(
        1, 64, kernel_size=(7, 7), stride=(2, 2), padding=(3, 3), bias=False
    )
    # [1] Prepare model.
    model = ray.train.torch.prepare_model(model)
    criterion = CrossEntropyLoss()
    optimizer = Adam(model.parameters(), lr=0.001)

    # Data
    transform = Compose([ToTensor(), Normalize((0.28604,), (0.32025,))])
    data_dir = os.path.join(tempfile.gettempdir(), "data")
    train_data = FashionMNIST(root=data_dir, train=True, download=True, transform=transform)
    train_loader = DataLoader(train_data, batch_size=128, shuffle=True)
    # [2] Prepare dataloader.
    train_loader = ray.train.torch.prepare_data_loader(train_loader)

    # Training
    for epoch in range(10):
        if ray.train.get_context().get_world_size() > 1:
            train_loader.sampler.set_epoch(epoch)

        for images, labels in train_loader:
            outputs = model(images)
            loss = criterion(outputs, labels)
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()

        # [3] Report metrics and checkpoint.
        metrics = {"loss": loss.item(), "epoch": epoch}
        with tempfile.TemporaryDirectory() as temp_checkpoint_dir:
            torch.save(
                model.state_dict(),
                os.path.join(temp_checkpoint_dir, "model.pt")
            )
            ray.train.report(
                metrics,
                checkpoint=ray.train.Checkpoint.from_directory(temp_checkpoint_dir),
            )
        if ray.train.get_context().get_world_rank() == 0:
            print(metrics)

# Configuration for local mode
use_gpu = True
scaling_config = ScalingConfig(num_workers=0, use_gpu=use_gpu)  # Local mode
run_config = RunConfig(checkpoint_config=CheckpointConfig(num_to_keep=1))

# Note: Ray Data not supported with multiple processes in local mode
# For multi-process training, use PyTorch DataLoader as shown above

# Initialize the Trainer
trainer = TorchTrainer(
    train_loop_per_worker=train_func,
    scaling_config=scaling_config,
    run_config=run_config,
)

# Train the model
result = trainer.fit()

Running Options:

# Option 1: Single process local mode
RAY_TRAIN_V2_ENABLED=1 python test.py

# Option 2: Multi-process local mode with torchrun
RAY_TRAIN_V2_ENABLED=1 torchrun --standalone --nnodes=1 --nproc-per-node=4 test.py

# Option 3: Switch to distributed Ray Train (change num_workers=4)
# Same training code works across all modes!

Checks

• I've signed off every commit(by using the -s flag, i.e., git commit -s) in this PR.
• I've run scripts/format.sh to lint the changes in this PR.
• I've included any doc changes needed for https://docs.ray.io/en/master/.
  • I've added any new APIs to the API Reference. For example, if I added a
    method in Tune, I've added it in doc/source/tune/api/ under the
    corresponding .rst file.
• I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
• Testing Strategy
  • Unit tests
  • Release tests
  • This PR is not tested :(

[liulehui]

nit: is_torch_distributed_env_vars_set() to align with https://github.com/ray-project/ray/blob/master/python/ray/train/torch/config.py#L143 ?

is the CUDA_VISIBLE_DEVICES env var needed to be set for cuda env?

[xinyuangui2]

Good call. Let me change to align.

[xinyuangui2]

After second thought, I think current env variables are the minimum requirement for processes. to communicate with each other.

[matthewdeng]

I don't think this is actually correct. In the example in the PR description, shouldn't LOCAL_WORLD_SIZE be 4?

# Option 2: Multi-process local mode with torchrun
RAY_TRAIN_V2_ENABLED=1 torchrun --standalone --nnodes=1 --nproc-per-node=4 test.py

[xinyuangui2]

Hmm, isn't LOCAL_WORLD_SIZE == world_size == 4?

[matthewdeng]

Oh you are right... I mixed it up.

Do we want to support multiple nodes?

E.g.

RAY_TRAIN_V2_ENABLED=1 torchrun --standalone --nnodes=4 --nproc-per-node=1 test.py

[matthewdeng]

I see in the comment it does say "Local mode only supports 1 node", but is this intentional? Why not support multiple?

[xinyuangui2]

Good point. Actually we can support multi nodes.

[matthewdeng]

nit: I don't think this is actually needed for torchrun because the process gets shut down. Can perhaps remove it for now and add back if we identify issues without it.

[xinyuangui2]

This is not needed in real run. But it is needed in pytest, where different tests run in the same process.

[matthewdeng]

Mostly lgtm, add tests?

[xinyuangui2]

Mostly lgtm, add tests?

For the torch change, it is already captured inside the test for torch trainer inside test_local_mode.

For the multi-processes change, it is hard to test inside unittest. So I am planning to add release tests.

[matthewdeng]