mirror of
https://github.com/wassname/pytorch-lightning.git
synced 2026-09-09 11:32:07 +08:00
cleaning up demos (#313)
* cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos * cleaning up demos
This commit is contained in:
William Falcon
@@ -51,14 +51,14 @@ if __name__ == '__main__':
|
||||
# gpu args
|
||||
parent_parser.add_argument(
|
||||
'--gpus',
|
||||
type=str,
|
||||
default='-1',
|
||||
help='any integer (number of GPUs to use) or -1 for all'
|
||||
type=int,
|
||||
default=2,
|
||||
help='how many gpus'
|
||||
)
|
||||
parent_parser.add_argument(
|
||||
'--distributed_backend',
|
||||
type=str,
|
||||
default=None,
|
||||
default='dp',
|
||||
help='supports three options dp, ddp, ddp2'
|
||||
)
|
||||
parent_parser.add_argument(
|
||||
|
||||
@@ -8,7 +8,7 @@ from torchvision.datasets import MNIST
|
||||
import torchvision.transforms as transforms
|
||||
import torch
|
||||
import torch.nn.functional as F
|
||||
from test_tube import HyperOptArgumentParser
|
||||
from argparse import ArgumentParser
|
||||
from torch import optim
|
||||
from torch.utils.data import DataLoader
|
||||
from torch.utils.data.distributed import DistributedSampler
|
||||
@@ -95,7 +95,7 @@ class LightningTemplateModel(LightningModule):
|
||||
loss_val = self.loss(y, y_hat)
|
||||
|
||||
# in DP mode (default) make sure if result is scalar, there's another dim in the beginning
|
||||
if self.trainer.use_dp:
|
||||
if self.trainer.use_dp or self.trainer.use_ddp2:
|
||||
loss_val = loss_val.unsqueeze(0)
|
||||
|
||||
output = OrderedDict({
|
||||
@@ -126,7 +126,7 @@ class LightningTemplateModel(LightningModule):
|
||||
val_acc = val_acc.cuda(loss_val.device.index)
|
||||
|
||||
# in DP mode (default) make sure if result is scalar, there's another dim in the beginning
|
||||
if self.trainer.use_dp:
|
||||
if self.trainer.use_dp or self.trainer.use_ddp2:
|
||||
loss_val = loss_val.unsqueeze(0)
|
||||
val_acc = val_acc.unsqueeze(0)
|
||||
|
||||
@@ -168,7 +168,7 @@ class LightningTemplateModel(LightningModule):
|
||||
val_loss_mean /= len(outputs)
|
||||
val_acc_mean /= len(outputs)
|
||||
tqdm_dict = {'val_loss': val_loss_mean, 'val_acc': val_acc_mean}
|
||||
result = {'progress_bar': tqdm_dict}
|
||||
result = {'progress_bar': tqdm_dict, 'logs': tqdm_dict}
|
||||
return result
|
||||
|
||||
# ---------------------
|
||||
@@ -190,20 +190,20 @@ class LightningTemplateModel(LightningModule):
|
||||
dataset = MNIST(root=self.hparams.data_root, train=train,
|
||||
transform=transform, download=True)
|
||||
|
||||
# when using multi-node (ddp) we need to add the datasampler
|
||||
# when using multi-node (ddp) we need to add the datasampler
|
||||
train_sampler = None
|
||||
batch_size = self.hparams.batch_size
|
||||
|
||||
if self.use_ddp:
|
||||
train_sampler = DistributedSampler(dataset, rank=self.trainer.proc_rank)
|
||||
batch_size = batch_size // self.trainer.world_size # scale batch size
|
||||
train_sampler = DistributedSampler(dataset)
|
||||
|
||||
should_shuffle = train_sampler is None
|
||||
loader = DataLoader(
|
||||
dataset=dataset,
|
||||
batch_size=batch_size,
|
||||
shuffle=should_shuffle,
|
||||
sampler=train_sampler
|
||||
sampler=train_sampler,
|
||||
num_workers=0
|
||||
)
|
||||
|
||||
return loader
|
||||
@@ -231,7 +231,7 @@ class LightningTemplateModel(LightningModule):
|
||||
:param root_dir:
|
||||
:return:
|
||||
"""
|
||||
parser = HyperOptArgumentParser(strategy=parent_parser.strategy, parents=[parent_parser])
|
||||
parser = ArgumentParser(parents=[parent_parser])
|
||||
|
||||
# param overwrites
|
||||
# parser.set_defaults(gradient_clip_val=5.0)
|
||||
@@ -241,21 +241,13 @@ class LightningTemplateModel(LightningModule):
|
||||
parser.add_argument('--out_features', default=10, type=int)
|
||||
# use 500 for CPU, 50000 for GPU to see speed difference
|
||||
parser.add_argument('--hidden_dim', default=50000, type=int)
|
||||
parser.opt_list('--drop_prob', default=0.2, options=[0.2, 0.5], type=float, tunable=True)
|
||||
parser.opt_list('--learning_rate', default=0.001 * 8, type=float,
|
||||
options=[0.0001, 0.0005, 0.001],
|
||||
tunable=True)
|
||||
parser.add_argument('--drop_prob', default=0.2, type=float)
|
||||
parser.add_argument('--learning_rate', default=0.001, type=float)
|
||||
|
||||
# data
|
||||
parser.add_argument('--data_root', default=os.path.join(root_dir, 'mnist'), type=str)
|
||||
|
||||
# training params (opt)
|
||||
parser.opt_list('--optimizer_name', default='adam', type=str,
|
||||
options=['adam'], tunable=False)
|
||||
|
||||
# if using 2 nodes with 4 gpus each the batch size here
|
||||
# (256) will be 256 / (2*8) = 16 per gpu
|
||||
parser.opt_list('--batch_size', default=256 * 8, type=int,
|
||||
options=[32, 64, 128, 256], tunable=False,
|
||||
help='batch size will be divided over all gpus being used across all nodes')
|
||||
parser.add_argument('--optimizer_name', default='adam', type=str)
|
||||
parser.add_argument('--batch_size', default=64, type=int)
|
||||
return parser
|
||||
|
||||
@@ -4,7 +4,17 @@ To run this demo which launches a single job that trains on 2 nodes (2 gpus per
|
||||
|
||||
1. Log into the jumphost node of your SLURM-managed cluster.
|
||||
2. Create a conda environment with Lightning and a GPU PyTorch version.
|
||||
3. Submit this script.
|
||||
3. Choose a script to submit
|
||||
|
||||
#### DDP
|
||||
Submit this job to run with distributedDataParallel (2 nodes, 2 gpus each)
|
||||
```bash
|
||||
sbatch job_submit.sh --env=YourEnv
|
||||
sbatch ddp_job_submit.sh YourEnv
|
||||
```
|
||||
|
||||
#### DDP2
|
||||
Submit this job to run with a different implementation of distributedDataParallel.
|
||||
In this version, each node acts like DataParallel but syncs across nodes like DDP.
|
||||
```bash
|
||||
sbatch ddp2_job_submit.sh YourEnv
|
||||
```
|
||||
|
||||
+27
@@ -0,0 +1,27 @@
|
||||
#!/bin/bash -l
|
||||
|
||||
# SLURM SUBMIT SCRIPT
|
||||
#SBATCH --nodes=2
|
||||
#SBATCH --gres=gpu:2
|
||||
#SBATCH --ntasks-per-node=1
|
||||
#SBATCH --mem=0
|
||||
#SBATCH --time=0-02:00:00
|
||||
|
||||
# activate conda env
|
||||
source activate $1
|
||||
|
||||
# -------------------------
|
||||
# debugging flags (optional)
|
||||
export NCCL_DEBUG=INFO
|
||||
export PYTHONFAULTHANDLER=1
|
||||
|
||||
# on your cluster you might need these:
|
||||
# set the network interface
|
||||
# export NCCL_SOCKET_IFNAME=^docker0,lo
|
||||
|
||||
# might need the latest cuda
|
||||
# module load NCCL/2.4.7-1-cuda.10.0
|
||||
# -------------------------
|
||||
|
||||
# run script from above
|
||||
srun python3 multi_node_ddp2_demo.py
|
||||
+4
-4
@@ -8,12 +8,12 @@
|
||||
#SBATCH --time=0-02:00:00
|
||||
|
||||
# activate conda env
|
||||
source activate $env
|
||||
source activate $1
|
||||
|
||||
# -------------------------
|
||||
# debugging flags (optional)
|
||||
# export NCCL_DEBUG=INFO
|
||||
# export PYTHONFAULTHANDLER=1
|
||||
export NCCL_DEBUG=INFO
|
||||
export PYTHONFAULTHANDLER=1
|
||||
|
||||
# on your cluster you might need these:
|
||||
# set the network interface
|
||||
@@ -24,4 +24,4 @@ source activate $env
|
||||
# -------------------------
|
||||
|
||||
# run script from above
|
||||
srun python multi_node_demo.py
|
||||
srun python3 multi_node_ddp_demo.py
|
||||
@@ -0,0 +1,55 @@
|
||||
"""
|
||||
Multi-node example (GPU)
|
||||
"""
|
||||
import os
|
||||
import numpy as np
|
||||
import torch
|
||||
|
||||
from argparse import ArgumentParser
|
||||
from pytorch_lightning import Trainer
|
||||
from examples.basic_examples.lightning_module_template import LightningTemplateModel
|
||||
|
||||
SEED = 2334
|
||||
torch.manual_seed(SEED)
|
||||
np.random.seed(SEED)
|
||||
|
||||
|
||||
def main(hparams):
|
||||
"""
|
||||
Main training routine specific for this project
|
||||
:param hparams:
|
||||
:return:
|
||||
"""
|
||||
# ------------------------
|
||||
# 1 INIT LIGHTNING MODEL
|
||||
# ------------------------
|
||||
model = LightningTemplateModel(hparams)
|
||||
|
||||
# ------------------------
|
||||
# 2 INIT TRAINER
|
||||
# ------------------------
|
||||
trainer = Trainer(
|
||||
gpus=2,
|
||||
nb_gpu_nodes=2,
|
||||
distributed_backend='ddp2'
|
||||
)
|
||||
|
||||
# ------------------------
|
||||
# 3 START TRAINING
|
||||
# ------------------------
|
||||
trainer.fit(model)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
root_dir = os.path.dirname(os.path.realpath(__file__))
|
||||
parent_parser = ArgumentParser(add_help=False)
|
||||
|
||||
# each LightningModule defines arguments relevant to it
|
||||
parser = LightningTemplateModel.add_model_specific_args(parent_parser, root_dir)
|
||||
hyperparams = parser.parse_args()
|
||||
|
||||
# ---------------------
|
||||
# RUN TRAINING
|
||||
# ---------------------
|
||||
main(hyperparams)
|
||||
+2
-1
@@ -30,7 +30,8 @@ def main(hparams):
|
||||
# ------------------------
|
||||
trainer = Trainer(
|
||||
gpus=2,
|
||||
nb_gpu_nodes=2
|
||||
nb_gpu_nodes=2,
|
||||
distributed_backend='ddp'
|
||||
)
|
||||
|
||||
# ------------------------
|
||||
@@ -35,7 +35,7 @@ class ModelSummary(object):
|
||||
out_sizes = []
|
||||
input_ = self.model.example_input_array
|
||||
|
||||
if self.model.use_ddp or self.model.use_dp or self.model.single_gpu:
|
||||
if self.model.on_gpu:
|
||||
input_ = input_.cuda(0)
|
||||
|
||||
if self.model.trainer.use_amp:
|
||||
|
||||
@@ -436,7 +436,7 @@ class Trainer(TrainerIO):
|
||||
|
||||
@property
|
||||
def data_parallel(self):
|
||||
return self.use_dp or self.use_ddp
|
||||
return self.use_dp or self.use_ddp or self.use_ddp2
|
||||
|
||||
def __determine_data_use_amount(self, train_percent_check, val_percent_check,
|
||||
test_percent_check, overfit_pct):
|
||||
@@ -536,7 +536,7 @@ class Trainer(TrainerIO):
|
||||
args.append(dataloader_idx)
|
||||
|
||||
# handle DP, DDP forward
|
||||
if self.use_ddp or self.use_dp:
|
||||
if self.use_ddp or self.use_dp or self.use_ddp2:
|
||||
output = model(*args)
|
||||
return output
|
||||
|
||||
@@ -636,83 +636,94 @@ class Trainer(TrainerIO):
|
||||
self.get_test_dataloaders = model.test_dataloader
|
||||
self.get_val_dataloaders = model.val_dataloader
|
||||
|
||||
if self.use_ddp and not isinstance(self.get_train_dataloader().sampler, DistributedSampler):
|
||||
msg = """
|
||||
You're using multiple gpus and multiple nodes without using a DistributedSampler
|
||||
to assign a subset of your data to each process. To silence this warning, pass a
|
||||
DistributedSampler to your DataLoader.
|
||||
# call warnings from proc zero only which triggers dataloaders
|
||||
# if those have to download data it will only happen on proc 0
|
||||
if self.proc_rank == 0:
|
||||
if self.use_ddp or self.use_ddp2 and not isinstance(self.get_train_dataloader().sampler, DistributedSampler):
|
||||
msg = """
|
||||
You're using multiple gpus and multiple nodes without using a DistributedSampler
|
||||
to assign a subset of your data to each process. To silence this warning, pass a
|
||||
DistributedSampler to your DataLoader.
|
||||
|
||||
ie: this:
|
||||
dataset = myDataset()
|
||||
dataloader = Dataloader(dataset)
|
||||
ie: this:
|
||||
dataset = myDataset()
|
||||
dataloader = Dataloader(dataset)
|
||||
|
||||
becomes:
|
||||
dataset = myDataset()
|
||||
dist_sampler = torch.utils.data.distributed.DistributedSampler(dataset)
|
||||
dataloader = Dataloader(dataset, sampler=dist_sampler)
|
||||
becomes:
|
||||
dataset = myDataset()
|
||||
dist_sampler = torch.utils.data.distributed.DistributedSampler(dataset)
|
||||
dataloader = Dataloader(dataset, sampler=dist_sampler)
|
||||
|
||||
If you want each process to load the full dataset, ignore this warning.
|
||||
"""
|
||||
warnings.warn(msg)
|
||||
If you want each process to load the full dataset, ignore this warning.
|
||||
"""
|
||||
warnings.warn(msg)
|
||||
|
||||
if self.use_ddp and self.get_val_dataloaders is not None:
|
||||
for dataloader in self.get_val_dataloaders():
|
||||
if not isinstance(dataloader.sampler, DistributedSampler):
|
||||
msg = """
|
||||
Your val_dataloader(s) don't use DistributedSampler.
|
||||
You're using multiple gpus and multiple nodes without using a DistributedSampler
|
||||
to assign a subset of your data to each process. To silence this warning, pass a
|
||||
DistributedSampler to your DataLoader.
|
||||
if self.use_ddp or self.use_ddp2 and self.get_val_dataloaders is not None:
|
||||
for dataloader in self.get_val_dataloaders():
|
||||
if not isinstance(dataloader.sampler, DistributedSampler):
|
||||
msg = """
|
||||
Your val_dataloader(s) don't use DistributedSampler.
|
||||
You're using multiple gpus and multiple nodes without using a DistributedSampler
|
||||
to assign a subset of your data to each process. To silence this warning, pass a
|
||||
DistributedSampler to your DataLoader.
|
||||
|
||||
ie: this:
|
||||
dataset = myDataset()
|
||||
dataloader = Dataloader(dataset)
|
||||
ie: this:
|
||||
dataset = myDataset()
|
||||
dataloader = Dataloader(dataset)
|
||||
|
||||
becomes:
|
||||
dataset = myDataset()
|
||||
dist_sampler = torch.utils.data.distributed.DistributedSampler(dataset)
|
||||
dataloader = Dataloader(dataset, sampler=dist_sampler)
|
||||
becomes:
|
||||
dataset = myDataset()
|
||||
dist_sampler = torch.utils.data.distributed.DistributedSampler(dataset)
|
||||
dataloader = Dataloader(dataset, sampler=dist_sampler)
|
||||
|
||||
If you want each process to load the full dataset, ignore this warning.
|
||||
"""
|
||||
warnings.warn(msg)
|
||||
break
|
||||
If you want each process to load the full dataset, ignore this warning.
|
||||
"""
|
||||
warnings.warn(msg)
|
||||
break
|
||||
|
||||
if self.use_ddp and self.get_test_dataloaders is not None:
|
||||
for dataloader in self.get_test_dataloaders():
|
||||
if not isinstance(dataloader.sampler, DistributedSampler):
|
||||
msg = """
|
||||
Your test_dataloader(s) don't use DistributedSampler.
|
||||
You're using multiple gpus and multiple nodes without using a DistributedSampler
|
||||
to assign a subset of your data to each process. To silence this warning, pass a
|
||||
DistributedSampler to your DataLoader.
|
||||
if self.use_ddp or self.use_ddp2 and self.get_test_dataloaders is not None:
|
||||
for dataloader in self.get_test_dataloaders():
|
||||
if not isinstance(dataloader.sampler, DistributedSampler):
|
||||
msg = """
|
||||
Your test_dataloader(s) don't use DistributedSampler.
|
||||
You're using multiple gpus and multiple nodes without using a DistributedSampler
|
||||
to assign a subset of your data to each process. To silence this warning, pass a
|
||||
DistributedSampler to your DataLoader.
|
||||
|
||||
ie: this:
|
||||
dataset = myDataset()
|
||||
dataloader = Dataloader(dataset)
|
||||
ie: this:
|
||||
dataset = myDataset()
|
||||
dataloader = Dataloader(dataset)
|
||||
|
||||
becomes:
|
||||
dataset = myDataset()
|
||||
dist_sampler = torch.utils.data.distributed.DistributedSampler(dataset)
|
||||
dataloader = Dataloader(dataset, sampler=dist_sampler)
|
||||
becomes:
|
||||
dataset = myDataset()
|
||||
dist_sampler = torch.utils.data.distributed.DistributedSampler(dataset)
|
||||
dataloader = Dataloader(dataset, sampler=dist_sampler)
|
||||
|
||||
If you want each process to load the full dataset, ignore this warning.
|
||||
"""
|
||||
warnings.warn(msg)
|
||||
break
|
||||
If you want each process to load the full dataset, ignore this warning.
|
||||
"""
|
||||
warnings.warn(msg)
|
||||
break
|
||||
|
||||
if self.use_ddp or self.use_ddp2:
|
||||
# wait for all processes to catch up
|
||||
dist.barrier()
|
||||
|
||||
# load each dataloader
|
||||
self.get_train_dataloader()
|
||||
self.get_test_dataloaders()
|
||||
self.get_val_dataloaders()
|
||||
|
||||
# -----------------------------
|
||||
# MODEL TRAINING
|
||||
# -----------------------------
|
||||
def fit(self, model):
|
||||
# when using multi-node or DDP within a node start each module in a separate process
|
||||
if self.use_ddp:
|
||||
if self.use_ddp2:
|
||||
task = int(os.environ['SLURM_LOCALID'])
|
||||
self.ddp_train(task, model)
|
||||
|
||||
if self.use_ddp2:
|
||||
task = int(os.environ['SLURM_LOCALID'])
|
||||
self.ddp_train(task, model)
|
||||
|
||||
elif self.is_slurm_managing_tasks:
|
||||
elif self.use_ddp:
|
||||
if self.is_slurm_managing_tasks:
|
||||
task = int(os.environ['SLURM_LOCALID'])
|
||||
self.ddp_train(task, model)
|
||||
else:
|
||||
@@ -901,14 +912,14 @@ class Trainer(TrainerIO):
|
||||
default_port = default_port[-4:]
|
||||
|
||||
# all ports should be in the 10k+ range
|
||||
default_port = int(default_port) + 10000
|
||||
default_port = int(default_port) + 15000
|
||||
|
||||
except Exception as e:
|
||||
default_port = 12910
|
||||
|
||||
# if user gave a port number, use that one instead
|
||||
try:
|
||||
port = os.environ['MASTER_PORT']
|
||||
default_port = os.environ['MASTER_PORT']
|
||||
except Exception:
|
||||
os.environ['MASTER_PORT'] = str(default_port)
|
||||
|
||||
@@ -920,7 +931,6 @@ class Trainer(TrainerIO):
|
||||
|
||||
root_node = self.resolve_root_node_address(root_node)
|
||||
os.environ['MASTER_ADDR'] = root_node
|
||||
|
||||
dist.init_process_group("nccl", rank=self.proc_rank, world_size=self.world_size)
|
||||
|
||||
def resolve_root_node_address(self, root_node):
|
||||
@@ -1122,7 +1132,7 @@ class Trainer(TrainerIO):
|
||||
|
||||
# add gpu memory
|
||||
if self.on_gpu and self.log_gpu_memory:
|
||||
mem_map = memory.get_memory_profile()
|
||||
mem_map = memory.get_memory_profile(self.log_gpu_memory)
|
||||
metrics.update(mem_map)
|
||||
|
||||
# add norms
|
||||
@@ -1205,7 +1215,7 @@ class Trainer(TrainerIO):
|
||||
if len(self.optimizers) > 1:
|
||||
args.append(opt_idx)
|
||||
|
||||
if self.use_ddp:
|
||||
if self.use_ddp or self.use_ddp2:
|
||||
output = self.model(*args)
|
||||
elif self.use_dp:
|
||||
output = self.model(*args)
|
||||
|
||||
Reference in New Issue
Block a user