diff --git a/examples/basic_examples/gpu_template.py b/examples/basic_examples/gpu_template.py index b3f53609..8c65f88b 100644 --- a/examples/basic_examples/gpu_template.py +++ b/examples/basic_examples/gpu_template.py @@ -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( diff --git a/examples/basic_examples/lightning_module_template.py b/examples/basic_examples/lightning_module_template.py index 7188d804..88b8534c 100644 --- a/examples/basic_examples/lightning_module_template.py +++ b/examples/basic_examples/lightning_module_template.py @@ -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 diff --git a/examples/multi_node_examples/README.md b/examples/multi_node_examples/README.md index 0db9d462..047ce9bd 100644 --- a/examples/multi_node_examples/README.md +++ b/examples/multi_node_examples/README.md @@ -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 ``` diff --git a/examples/multi_node_examples/ddp2_job_submit.sh b/examples/multi_node_examples/ddp2_job_submit.sh new file mode 100755 index 00000000..6e433f5f --- /dev/null +++ b/examples/multi_node_examples/ddp2_job_submit.sh @@ -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 diff --git a/examples/multi_node_examples/job_submit.sh b/examples/multi_node_examples/ddp_job_submit.sh similarity index 80% rename from examples/multi_node_examples/job_submit.sh rename to examples/multi_node_examples/ddp_job_submit.sh index 176bf09f..bf53a653 100755 --- a/examples/multi_node_examples/job_submit.sh +++ b/examples/multi_node_examples/ddp_job_submit.sh @@ -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 diff --git a/examples/multi_node_examples/multi_node_ddp2_demo.py b/examples/multi_node_examples/multi_node_ddp2_demo.py new file mode 100644 index 00000000..244eb044 --- /dev/null +++ b/examples/multi_node_examples/multi_node_ddp2_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) diff --git a/examples/multi_node_examples/multi_node_demo.py b/examples/multi_node_examples/multi_node_ddp_demo.py similarity index 95% rename from examples/multi_node_examples/multi_node_demo.py rename to examples/multi_node_examples/multi_node_ddp_demo.py index 577336b8..c380497b 100644 --- a/examples/multi_node_examples/multi_node_demo.py +++ b/examples/multi_node_examples/multi_node_ddp_demo.py @@ -30,7 +30,8 @@ def main(hparams): # ------------------------ trainer = Trainer( gpus=2, - nb_gpu_nodes=2 + nb_gpu_nodes=2, + distributed_backend='ddp' ) # ------------------------ diff --git a/pytorch_lightning/root_module/memory.py b/pytorch_lightning/root_module/memory.py index 6aa4519e..ec4e2c02 100644 --- a/pytorch_lightning/root_module/memory.py +++ b/pytorch_lightning/root_module/memory.py @@ -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: diff --git a/pytorch_lightning/trainer/trainer.py b/pytorch_lightning/trainer/trainer.py index d908ec50..f79cdb1a 100644 --- a/pytorch_lightning/trainer/trainer.py +++ b/pytorch_lightning/trainer/trainer.py @@ -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)