From 146baa41e89c5521637da9477ff343c1f4fee6fe Mon Sep 17 00:00:00 2001 From: selee Date: Tue, 10 Feb 2026 16:10:10 +0900 Subject: [PATCH] Add ConfigMap training script, update IB/NCCL config in README - Add ConfigMap with train_nccl.py (VGG11 + CIFAR10 + NCCL) - Runtime: mount training-scripts ConfigMap to /workspace/scripts - Runtime: set command to python /workspace/scripts/train_nccl.py - Runtime: remove /dev/infiniband hostPath (hostdevice-net CNI handles it) - Runtime: add NCCL_DEBUG_SUBSYS=INIT,NET,IB - Runtime: add cifar-10 dataset hostPath mount - README: add NCCL_DEBUG_SUBSYS, cifar-data, MASTER_ADDR auto-config note Co-Authored-By: Claude Opus 4.6 --- README.md | 21 +- configs/volcano-trainjob-integration.yaml | 288 +++++++++++++++++++++- 2 files changed, 299 insertions(+), 10 deletions(-) diff --git a/README.md b/README.md index f0c1ef5..38dab78 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,8 @@ |------|-----| | Kubernetes | v1.34.3 | | Helm | v4.0.4 | -| GPU 노드 | InfiniBand 연결 GPU 노드 3대 | +| GPU 노드 | InfiniBand 연결 GPU 노드 3대 (노드당 GPU 8장, IB NIC 2개) | +| GPU 이미지 | nvcr.io/nvidia/pytorch:24.10-py3 | | Volcano | v1.14.0 | | Kubeflow Trainer | sha-48e7a93 | | JobSet (의존성) | v0.10.1 | @@ -185,6 +186,22 @@ Volcano 스케줄러를 Kubeflow TrainJob과 연동하여 gang scheduling, queue | `template.metadata.annotations` | Queue 지정 등. TrainJob level에서 override 가능 | | `resources.requests/limits` | 노드당 GPU 기본 할당량 (TrainJob에서 override 가능) | +현재 Runtime에 포함된 InfiniBand/NCCL 설정: + +| 설정 | 값 | 설명 | +|------|-----|------| +| `k8s.v1.cni.cncf.io/networks` | `hostdevice-net` | InfiniBand CNI 네트워크 연결 | +| `nvidia.com/gpu` | `8` | 노드당 GPU 8장 | +| `nvidia.com/hostdev` | `2` | 노드당 IB NIC 2개 | +| `privileged` + `IPC_LOCK` | - | NCCL IB 통신에 필요한 권한 | +| `NCCL_IB_DISABLE` | `0` | InfiniBand 사용 | +| `NCCL_DEBUG_SUBSYS` | `INIT,NET,IB` | NCCL 디버그 서브시스템 | +| `NCCL_SOCKET_IFNAME` | `net` | NCCL 소켓 인터페이스 | +| `/dev/shm` | `128Gi` | NCCL 공유 메모리 (GPU 8장 기준) | +| `cifar-data` | hostPath | `/home/ubuntu/cifar-10-batches-py` 데이터셋 마운트 | + +> MASTER_ADDR / MASTER_PORT는 Kubeflow Trainer torch runtime이 torchrun rdzv로 자동 설정하므로 별도 지정 불필요. + #### TrainJob — 학습 작업 제출 실제 학습을 실행할 때마다 생성합니다. Runtime의 기본값을 상속받고, 필요한 부분만 override합니다. @@ -197,7 +214,7 @@ Volcano 스케줄러를 Kubeflow TrainJob과 연동하여 gang scheduling, queue | `trainer.numProcPerNode` | 노드당 프로세스 수 (`"auto"`면 GPU 수만큼 자동) | | `trainer.resourcesPerNode` | 노드당 리소스 (Runtime 기본값 override) | -> 설치 테스트는 `numNodes: 1`로 단일노드에서 진행합니다. 검증 완료 후 `numNodes`를 늘리면 멀티노드 분산학습이 됩니다. +> 현재 기본 설정은 `numNodes: 2` (2노드 멀티노드 분산학습). 단일노드 테스트 시 TrainJob에서 `numNodes: 1`로 override 가능. ### 4.2 연동 설정 적용 diff --git a/configs/volcano-trainjob-integration.yaml b/configs/volcano-trainjob-integration.yaml index 416eae4..7279d3c 100644 --- a/configs/volcano-trainjob-integration.yaml +++ b/configs/volcano-trainjob-integration.yaml @@ -1,6 +1,268 @@ # Volcano <-> Kubeflow Trainer Integration Configuration Sample # Reference: https://www.kubeflow.org/docs/components/trainer/gang-scheduling/volcano/ +--- +# 0. ConfigMap for test training script (VGG11 + CIFAR10 + NCCL) +apiVersion: v1 +kind: ConfigMap +metadata: + name: training-scripts +data: + train_nccl.py: | + import os + import time + import torch + import argparse + import torch.distributed as dist + import torch.multiprocessing as mp + from torch.optim.lr_scheduler import StepLR + + # for dataset + from torchvision.datasets.cifar import CIFAR10 + import torchvision.transforms as tfs + from torch.utils.data import DataLoader + from torch.utils.data.distributed import DistributedSampler + + # for model + from torchvision.models import vgg11 + from torch.nn.parallel import DistributedDataParallel as DDP + + import numpy as np + import random + import datetime + + + def set_random_seeds(random_seed=0): + torch.manual_seed(random_seed) + torch.backends.cudnn.deterministic = True + torch.backends.cudnn.benchmark = False + np.random.seed(random_seed) + random.seed(random_seed) + + + def get_args_parser(): + parser = argparse.ArgumentParser(add_help=False) + parser.add_argument('--lr', type=float, default=0.01) + parser.add_argument('--epoch', type=int, default=90) + parser.add_argument('--batch_size', type=int, default=1200) + parser.add_argument('--global_rank', type=int, default=0) + parser.add_argument('--vis_step', type=int, default=10) + parser.add_argument('--num_workers', type=int, default=24) + parser.add_argument("--local_rank", type=int, + help="Local rank. Necessary for using the torch.distributed.launch utility.") + parser.add_argument('--world_size', type=int, default=0) + parser.add_argument('--port', type=int, default=2022) + parser.add_argument('--root', type=str, default='data') + parser.add_argument('--start_epoch', type=int, default=0) + parser.add_argument('--save_path', type=str, default='./save') + parser.add_argument('--save_file_name', type=str, default='vgg_cifar') + return parser + + + def main(opts): + # 1. set random seeds + set_random_seeds(random_seed=0) + + # 2. initialization + init_for_distributed(opts) + + # 3. visdom + vis = None + + # 4. data set + transform_train = tfs.Compose([ + tfs.Resize(256), + tfs.RandomCrop(224), + tfs.RandomHorizontalFlip(), + tfs.ToTensor(), + tfs.Normalize(mean=(0.4914, 0.4822, 0.4465), + std=(0.2023, 0.1994, 0.2010)), + ]) + + transform_test = tfs.Compose([ + tfs.Resize(256), + tfs.CenterCrop(224), + tfs.ToTensor(), + tfs.Normalize(mean=(0.4914, 0.4822, 0.4465), + std=(0.2023, 0.1994, 0.2010)), + ]) + + train_set = CIFAR10(root=opts.root, + train=True, + transform=transform_train, + download=True) + + test_set = CIFAR10(root=opts.root, + train=False, + transform=transform_test, + download=True) + + train_sampler = DistributedSampler(dataset=train_set, shuffle=True) + test_sampler = DistributedSampler(dataset=test_set, shuffle=False) + + train_loader = DataLoader(dataset=train_set, + batch_size=int(opts.batch_size / opts.world_size), + shuffle=False, + num_workers=int(opts.num_workers / opts.world_size), + sampler=train_sampler, + pin_memory=True) + + test_loader = DataLoader(dataset=test_set, + batch_size=int(opts.batch_size / opts.world_size), + shuffle=False, + num_workers=int(opts.num_workers / opts.world_size), + sampler=test_sampler, + pin_memory=True) + + # 5. model + model = vgg11(pretrained=False) + model = model.cuda(opts.local_rank) + model = DDP(module=model, + device_ids=[opts.local_rank]) + + # 6. criterion + criterion = torch.nn.CrossEntropyLoss().to(opts.local_rank) + + # 7. optimizer + optimizer = torch.optim.SGD(params=model.parameters(), + lr=0.01, + weight_decay=0.0005, + momentum=0.9) + + # 8. scheduler + scheduler = StepLR(optimizer=optimizer, + step_size=30, + gamma=0.1) + + if opts.start_epoch != 0: + + checkpoint = torch.load(os.path.join(opts.save_path, opts.save_file_name) + '.{}.pth.tar' + .format(opts.start_epoch - 1), + map_location=torch.device('cuda:{}'.format(opts.local_rank))) + model.load_state_dict(checkpoint['model_state_dict']) + optimizer.load_state_dict(checkpoint['optimizer_state_dict']) + scheduler.load_state_dict(checkpoint['scheduler_state_dict']) + if opts.global_rank == 0: + print('\nLoaded checkpoint from epoch %d.\n' % (int(opts.start_epoch) - 1)) + + for epoch in range(opts.start_epoch, opts.epoch): + + # 9. train + tic = time.time() + model.train() + train_sampler.set_epoch(epoch) + + for i, (images, labels) in enumerate(train_loader): + images = images.to(opts.local_rank) + labels = labels.to(opts.local_rank) + outputs = model(images) + + # ----------- update ----------- + optimizer.zero_grad() + loss = criterion(outputs, labels) + loss.backward() + optimizer.step() + + # get lr + for param_group in optimizer.param_groups: + lr = param_group['lr'] + + # time + toc = time.time() + + # visualization + if (i % opts.vis_step == 0 or i == len(train_loader) - 1): + print('GPU[{0}] Epoch [{1}/{2}], Iter [{3}/{4}], Loss: {5:.4f}, LR: {6:.5f}, Time: {7:.2f}'.format(opts.global_rank, + epoch, + opts.epoch, + i, + len(train_loader), + loss.item(), + lr, + toc - tic)) + + # save pth file + if opts.local_rank == 0: + if not os.path.exists(opts.save_path): + os.mkdir(opts.save_path) + + checkpoint = {'epoch': epoch, + 'model_state_dict': model.state_dict(), + 'optimizer_state_dict': optimizer.state_dict(), + 'scheduler_state_dict': scheduler.state_dict()} + + torch.save(checkpoint, os.path.join(opts.save_path, opts.save_file_name + '.{}.pth.tar'.format(epoch))) + print("save pth.tar {} epoch!".format(epoch)) + + # 10. test + model.eval() + + val_avg_loss = 0 + correct_top1 = 0 + correct_top5 = 0 + total = 0 + + with torch.no_grad(): + for i, (images, labels) in enumerate(test_loader): + images = images.to(opts.local_rank) + labels = labels.to(opts.local_rank) + outputs = model(images) + loss = criterion(outputs, labels) + val_avg_loss += loss.item() + # rank 1 + _, pred = torch.max(outputs, 1) + total += labels.size(0) + correct_top1 += (pred == labels).sum().item() + + # rank 5 + _, rank5 = outputs.topk(5, 1, True, True) + rank5 = rank5.t() + correct5 = rank5.eq(labels.view(1, -1).expand_as(rank5)) + + for k in range(5): + correct_k = correct5[:k+1].reshape(-1).float().sum(0, keepdim=True) + correct_top5 += correct_k.item() + + accuracy_top1 = correct_top1 / total + accuracy_top5 = correct_top5 / total + + val_avg_loss = val_avg_loss / len(test_loader) + + print("top-1 percentage : {0:0.3f}%".format(correct_top1 / total * 100)) + print("top-5 percentage : {0:0.3f}%".format(correct_top5 / total * 100)) + scheduler.step() + + return 0 + + + def init_for_distributed(opts): + + # 1. setting for distributed training + opts.global_rank = int(os.environ['RANK']) + opts.local_rank = int(os.environ['LOCAL_RANK']) + opts.world_size = int(os.environ['WORLD_SIZE']) + torch.cuda.set_device(opts.local_rank) + if opts.global_rank is not None and opts.local_rank is not None: + print("Use GPU: [{}/{}] for training".format(opts.global_rank, opts.local_rank)) + + # 2. init_process_group + dist.init_process_group( + backend="nccl", + rank=opts.global_rank, + world_size=opts.world_size, + device_id=torch.device(f"cuda:{opts.local_rank}"), + timeout=datetime.timedelta(seconds=60) + ) + + return + + + if __name__ == '__main__': + + parser = argparse.ArgumentParser('vgg11 cifar training', parents=[get_args_parser()]) + opts = parser.parse_args() + main(opts) + --- # 1. Volcano Queue for training workloads # Queue는 리소스 할당 단위. capability는 클러스터 실제 용량에 맞게 조정 필요. @@ -21,6 +283,7 @@ spec: --- # 2. ClusterTrainingRuntime with Volcano gang scheduling + topology-aware scheduling # podGroupPolicy.volcano를 사용하면 PodGroup이 자동 생성됨 (수동 생성 불필요) +# MASTER_ADDR / MASTER_PORT는 Kubeflow Trainer torch runtime이 자동 설정 (torchrun rdzv) apiVersion: trainer.kubeflow.org/v1alpha1 kind: ClusterTrainingRuntime metadata: @@ -57,6 +320,9 @@ spec: containers: - name: node image: nvcr.io/nvidia/pytorch:24.10-py3 + command: + - python + - /workspace/scripts/train_nccl.py securityContext: privileged: true capabilities: @@ -65,6 +331,8 @@ spec: env: - name: NCCL_DEBUG value: "INFO" + - name: NCCL_DEBUG_SUBSYS + value: "INIT,NET,IB" - name: NCCL_IB_DISABLE value: "0" - name: NCCL_SOCKET_IFNAME @@ -79,22 +347,27 @@ spec: volumeMounts: - name: shared-memory mountPath: /dev/shm - - name: infiniband - mountPath: /dev/infiniband + - name: cifar-data + mountPath: /workspace/data/cifar-10-batches-py + - name: training-scripts + mountPath: /workspace/scripts volumes: - name: shared-memory emptyDir: medium: Memory sizeLimit: 128Gi - - name: infiniband + - name: cifar-data hostPath: - path: /dev/infiniband + path: /home/ubuntu/cifar-10-batches-py type: Directory + - name: training-scripts + configMap: + name: training-scripts + defaultMode: 0755 --- -# 3. Example TrainJob (설치 테스트용 - 단일노드) -# podGroupPolicy는 runtime에서 상속되므로 TrainJob에서는 별도 설정 불필요 -# 설치 테스트는 numNodes: 1로 단일노드에서 진행. 멀티노드는 검증 후 numNodes를 늘리면 됨. +# 3. Example TrainJob (테스트용 - 2노드 분산학습) +# podGroupPolicy, env, volumes, command 등은 runtime에서 상속됨 apiVersion: trainer.kubeflow.org/v1alpha1 kind: TrainJob metadata: @@ -104,7 +377,6 @@ spec: runtimeRef: name: torch-distributed-volcano trainer: - # TODO: Replace with your actual training image image: nvcr.io/nvidia/pytorch:24.10-py3 numNodes: 2 resourcesPerNode: