mirror of
https://github.com/wassname/pytorch-lightning.git
synced 2026-09-12 12:40:20 +08:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c2e2298586 | ||
|
|
319feb7da5 | ||
|
|
5195124d4e | ||
|
|
4e67983f23 | ||
|
|
c02b6c4c88 | ||
|
|
ad44d9168b | ||
|
|
53a1a6d462 | ||
|
|
59d60eaf18 | ||
|
|
112be99b19 | ||
|
|
0e67773d2e | ||
|
|
394cdeeb8b | ||
|
|
d0a8292e02 | ||
|
|
d7409afed9 | ||
|
|
f01cb63234 | ||
|
|
8be7480f31 | ||
|
|
751bc7c695 | ||
|
|
3be26dbb95 | ||
|
|
2ca0864ce8 | ||
|
|
b1041220ac | ||
|
|
da842c0cd6 | ||
|
|
c4971e8432 | ||
|
|
e81dbce38c | ||
|
|
0d992689d5 | ||
|
|
4085b3fa69 | ||
|
|
b684bb55c5 | ||
|
|
f98f88ff08 | ||
|
|
f0955df4f0 | ||
|
|
7744c7117d | ||
|
|
22f4d6e26e | ||
|
|
d49a83dec0 | ||
|
|
6d1d5ef68e | ||
|
|
3a1525222d | ||
|
|
f650253cae | ||
|
|
c67c84b443 | ||
|
|
4db32984c6 | ||
|
|
81d39786d9 | ||
|
|
63de076765 | ||
|
|
256ca62a3c | ||
|
|
39d04eb795 | ||
|
|
e02857fcce | ||
|
|
c163caf8cb | ||
|
|
2096a0aa84 | ||
|
|
c253f96c53 | ||
|
|
551daca047 | ||
|
|
ded0abead7 | ||
|
|
e86b191691 | ||
|
|
3321e8c541 | ||
|
|
bc3a805202 | ||
|
|
162b9f4f27 | ||
|
|
e5bc3ea5b4 | ||
|
|
baa139f97a | ||
|
|
470f3e6d29 | ||
|
|
c12a0b57da | ||
|
|
e7ecfa15f8 | ||
|
|
9051eb0039 | ||
|
|
bb8dbfca09 | ||
|
|
0240c70780 | ||
|
|
a41abad5b2 | ||
|
|
a83588b14e | ||
|
|
80192752b7 | ||
|
|
fbd3873a0f | ||
|
|
28cfddbe65 | ||
|
|
b4bdb283ce | ||
|
|
967e57f071 | ||
|
|
d12f6b7dd8 | ||
|
|
182c025c88 | ||
|
|
58e6199ce8 | ||
|
|
6a33f0d483 | ||
|
|
dd230a93e8 | ||
|
|
3aa9cfc18e | ||
|
|
e57f461323 | ||
|
|
ab00514ef6 | ||
|
|
1dd58b4687 | ||
|
|
b4b8a3dfde | ||
|
|
d8782c7b90 |
@@ -116,8 +116,8 @@ def validation_end(self, outputs):
|
||||
return tqdm_dic
|
||||
```
|
||||
|
||||
## TensorboardX
|
||||
Lightning is fully integrated with tensorboardX.
|
||||
## Tensorboard
|
||||
Lightning is fully integrated with tensorboard.
|
||||
|
||||
<p align="center">
|
||||
<a href="https://williamfalcon.github.io/pytorch-lightning/">
|
||||
|
||||
@@ -19,7 +19,7 @@ Cut the learning rate by 10 at every epoch listed in this list.
|
||||
trainer = Trainer(lr_scheduler_milestones=None)
|
||||
|
||||
# cut LR by 10 at 100, 200, and 300 epochs
|
||||
trainer = Trainer(lr_scheduler_milestones=[100, 200, 300])
|
||||
trainer = Trainer(lr_scheduler_milestones='100, 200, 300')
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
@@ -84,9 +84,10 @@ class LightningTemplateModel(LightningModule):
|
||||
loss_val = self.loss(y, y_hat)
|
||||
|
||||
output = OrderedDict({
|
||||
'loss': loss_val,
|
||||
'tqdm_metrics': {}
|
||||
'loss': loss_val
|
||||
})
|
||||
|
||||
# can also return just a scalar instead of a dict (return loss_val)
|
||||
return output
|
||||
|
||||
def validation_step(self, data_batch, batch_i):
|
||||
@@ -107,8 +108,10 @@ class LightningTemplateModel(LightningModule):
|
||||
|
||||
output = OrderedDict({
|
||||
'val_loss': loss_val,
|
||||
'val_acc': torch.tensor(val_acc),
|
||||
'val_acc': torch.tensor(val_acc).cuda(loss_val.device.index),
|
||||
})
|
||||
|
||||
# can also return just a scalar instead of a dict (return loss_val)
|
||||
return output
|
||||
|
||||
def validation_end(self, outputs):
|
||||
@@ -117,6 +120,10 @@ class LightningTemplateModel(LightningModule):
|
||||
:param outputs: list of individual outputs of each validation step
|
||||
:return:
|
||||
"""
|
||||
# if returned a scalar from validation_step, outputs is a list of tensor scalars
|
||||
# we return just the average in this case (if we want)
|
||||
# return torch.stack(outputs).mean()
|
||||
|
||||
val_loss_mean = 0
|
||||
val_acc_mean = 0
|
||||
for output in outputs:
|
||||
|
||||
@@ -0,0 +1,112 @@
|
||||
"""
|
||||
Runs a model on a single node across N-gpus.
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
import numpy as np
|
||||
from time import sleep
|
||||
import torch
|
||||
|
||||
from test_tube import HyperOptArgumentParser, Experiment, SlurmCluster
|
||||
from pytorch_lightning.models.trainer import Trainer
|
||||
from pytorch_lightning.utils.arg_parse import add_default_args
|
||||
|
||||
from pytorch_lightning.callbacks import EarlyStopping, ModelCheckpoint
|
||||
|
||||
SEED = 2334
|
||||
torch.manual_seed(SEED)
|
||||
np.random.seed(SEED)
|
||||
|
||||
from lightning_module_template import LightningTemplateModel
|
||||
|
||||
|
||||
def main(hparams):
|
||||
"""
|
||||
Main training routine specific for this project
|
||||
:param hparams:
|
||||
:return:
|
||||
"""
|
||||
# ------------------------
|
||||
# 1 INIT LIGHTNING MODEL
|
||||
# ------------------------
|
||||
print('loading model...')
|
||||
model = LightningTemplateModel(hparams)
|
||||
print('model built')
|
||||
|
||||
# ------------------------
|
||||
# 2 INIT TEST TUBE EXP
|
||||
# ------------------------
|
||||
|
||||
# init experiment
|
||||
exp = Experiment(
|
||||
name=hyperparams.experiment_name,
|
||||
save_dir=hyperparams.test_tube_save_path,
|
||||
autosave=False,
|
||||
description='test demo'
|
||||
)
|
||||
|
||||
exp.argparse(hparams)
|
||||
exp.save()
|
||||
|
||||
# ------------------------
|
||||
# 3 DEFINE CALLBACKS
|
||||
# ------------------------
|
||||
model_save_path = '{}/{}/{}'.format(hparams.model_save_path, exp.name, exp.version)
|
||||
early_stop = EarlyStopping(
|
||||
monitor='val_acc',
|
||||
patience=3,
|
||||
verbose=True,
|
||||
mode='max'
|
||||
)
|
||||
|
||||
checkpoint = ModelCheckpoint(
|
||||
filepath=model_save_path,
|
||||
save_best_only=True,
|
||||
verbose=True,
|
||||
monitor='val_loss',
|
||||
mode='min'
|
||||
)
|
||||
|
||||
# ------------------------
|
||||
# 4 INIT TRAINER
|
||||
# ------------------------
|
||||
trainer = Trainer(
|
||||
experiment=exp,
|
||||
checkpoint_callback=checkpoint,
|
||||
early_stop_callback=early_stop,
|
||||
gpus=hparams.gpus,
|
||||
)
|
||||
|
||||
# ------------------------
|
||||
# 5 START TRAINING
|
||||
# ------------------------
|
||||
trainer.fit(model)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
# dirs
|
||||
root_dir = os.path.dirname(os.path.realpath(__file__))
|
||||
demo_log_dir = os.path.join(root_dir, 'pt_lightning_demo_logs')
|
||||
checkpoint_dir = os.path.join(demo_log_dir, 'model_weights')
|
||||
test_tube_dir = os.path.join(demo_log_dir, 'test_tube_data')
|
||||
|
||||
# although we user hyperOptParser, we are using it only as argparse right now
|
||||
parent_parser = HyperOptArgumentParser(strategy='grid_search', add_help=False)
|
||||
|
||||
# gpu args
|
||||
parent_parser.add_argument('--gpus', type=str, default='-1', help='how many gpus to use in the node. -1 uses all the gpus on the node')
|
||||
parent_parser.add_argument('--test_tube_save_path', type=str, default=test_tube_dir, help='where to save logs')
|
||||
parent_parser.add_argument('--model_save_path', type=str, default=checkpoint_dir, help='where to save model')
|
||||
parent_parser.add_argument('--experiment_name', type=str, default='pt_lightning_exp_a', help='test tube exp name')
|
||||
|
||||
# allow model to overwrite or extend args
|
||||
parser = LightningTemplateModel.add_model_specific_args(parent_parser, root_dir)
|
||||
hyperparams = parser.parse_args()
|
||||
|
||||
# ---------------------
|
||||
# RUN TRAINING
|
||||
# ---------------------
|
||||
# run on HPC cluster
|
||||
print(f'RUNNING INTERACTIVE MODE ON GPUS. gpu ids: {hyperparams.gpus}')
|
||||
main(hyperparams)
|
||||
@@ -0,0 +1,112 @@
|
||||
"""
|
||||
Runs a model on a single node across N-gpus.
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
import numpy as np
|
||||
from time import sleep
|
||||
import torch
|
||||
|
||||
from test_tube import HyperOptArgumentParser, Experiment, SlurmCluster
|
||||
from pytorch_lightning.models.trainer import Trainer
|
||||
from pytorch_lightning.utils.arg_parse import add_default_args
|
||||
|
||||
from pytorch_lightning.callbacks import EarlyStopping, ModelCheckpoint
|
||||
|
||||
SEED = 2334
|
||||
torch.manual_seed(SEED)
|
||||
np.random.seed(SEED)
|
||||
|
||||
from lightning_module_template import LightningTemplateModel
|
||||
|
||||
|
||||
def main(hparams):
|
||||
"""
|
||||
Main training routine specific for this project
|
||||
:param hparams:
|
||||
:return:
|
||||
"""
|
||||
# ------------------------
|
||||
# 1 INIT LIGHTNING MODEL
|
||||
# ------------------------
|
||||
print('loading model...')
|
||||
model = LightningTemplateModel(hparams)
|
||||
print('model built')
|
||||
|
||||
# ------------------------
|
||||
# 2 INIT TEST TUBE EXP
|
||||
# ------------------------
|
||||
|
||||
# init experiment
|
||||
exp = Experiment(
|
||||
name=hyperparams.experiment_name,
|
||||
save_dir=hyperparams.test_tube_save_path,
|
||||
autosave=False,
|
||||
description='test demo'
|
||||
)
|
||||
|
||||
exp.argparse(hparams)
|
||||
exp.save()
|
||||
|
||||
# ------------------------
|
||||
# 3 DEFINE CALLBACKS
|
||||
# ------------------------
|
||||
model_save_path = '{}/{}/{}'.format(hparams.model_save_path, exp.name, exp.version)
|
||||
early_stop = EarlyStopping(
|
||||
monitor='val_acc',
|
||||
patience=3,
|
||||
verbose=True,
|
||||
mode='max'
|
||||
)
|
||||
|
||||
checkpoint = ModelCheckpoint(
|
||||
filepath=model_save_path,
|
||||
save_best_only=True,
|
||||
verbose=True,
|
||||
monitor='val_loss',
|
||||
mode='min'
|
||||
)
|
||||
|
||||
# ------------------------
|
||||
# 4 INIT TRAINER
|
||||
# ------------------------
|
||||
trainer = Trainer(
|
||||
experiment=exp,
|
||||
checkpoint_callback=checkpoint,
|
||||
early_stop_callback=early_stop,
|
||||
gpus=hparams.gpus,
|
||||
)
|
||||
|
||||
# ------------------------
|
||||
# 5 START TRAINING
|
||||
# ------------------------
|
||||
trainer.fit(model)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
# dirs
|
||||
root_dir = os.path.dirname(os.path.realpath(__file__))
|
||||
demo_log_dir = os.path.join(root_dir, 'pt_lightning_demo_logs')
|
||||
checkpoint_dir = os.path.join(demo_log_dir, 'model_weights')
|
||||
test_tube_dir = os.path.join(demo_log_dir, 'test_tube_data')
|
||||
|
||||
# although we user hyperOptParser, we are using it only as argparse right now
|
||||
parent_parser = HyperOptArgumentParser(strategy='grid_search', add_help=False)
|
||||
|
||||
# gpu args
|
||||
parent_parser.add_argument('--gpus', type=str, default='0', help='how many gpus to use in the node. -1 uses all the gpus on the node')
|
||||
parent_parser.add_argument('--test_tube_save_path', type=str, default=test_tube_dir, help='where to save logs')
|
||||
parent_parser.add_argument('--model_save_path', type=str, default=checkpoint_dir, help='where to save model')
|
||||
parent_parser.add_argument('--experiment_name', type=str, default='pt_lightning_exp_a', help='test tube exp name')
|
||||
|
||||
# allow model to overwrite or extend args
|
||||
parser = LightningTemplateModel.add_model_specific_args(parent_parser, root_dir)
|
||||
hyperparams = parser.parse_args()
|
||||
|
||||
# ---------------------
|
||||
# RUN TRAINING
|
||||
# ---------------------
|
||||
# run on HPC cluster
|
||||
print(f'RUNNING INTERACTIVE MODE ON GPUS. gpu ids: {hyperparams.gpus}')
|
||||
main(hyperparams)
|
||||
@@ -6,6 +6,7 @@ import subprocess
|
||||
import traceback
|
||||
import warnings
|
||||
import os
|
||||
import pdb
|
||||
|
||||
import torch
|
||||
from torch.utils.data.distributed import DistributedSampler
|
||||
@@ -17,7 +18,7 @@ import tqdm
|
||||
|
||||
from pytorch_lightning.root_module.memory import get_gpu_memory_map
|
||||
from pytorch_lightning.root_module.model_saving import TrainerIO
|
||||
from pytorch_lightning.pt_overrides.override_data_parallel import LightningDistributedDataParallel
|
||||
from pytorch_lightning.pt_overrides.override_data_parallel import LightningDistributedDataParallel, LightningDataParallel
|
||||
|
||||
|
||||
try:
|
||||
@@ -27,11 +28,33 @@ except ModuleNotFoundError:
|
||||
APEX_AVAILABLE = False
|
||||
|
||||
|
||||
def reduce_distributed_output(output, nb_gpus):
|
||||
if nb_gpus <= 1:
|
||||
return output
|
||||
|
||||
# when using DP, we get one output per gpu
|
||||
# average outputs and return
|
||||
if type(output) is torch.Tensor:
|
||||
return output.mean()
|
||||
|
||||
for k, v in output.items():
|
||||
# recurse on nested dics
|
||||
if isinstance(output[k], dict):
|
||||
output[k] = reduce_distributed_output(output[k], nb_gpus)
|
||||
|
||||
# reduce only metrics that have the same nb of gpus
|
||||
elif output[k].size(0) == nb_gpus:
|
||||
reduced = torch.mean(output[k])
|
||||
output[k] = reduced
|
||||
return output
|
||||
|
||||
|
||||
class Trainer(TrainerIO):
|
||||
|
||||
def __init__(self,
|
||||
experiment,
|
||||
checkpoint_callback, early_stop_callback,
|
||||
early_stop_callback=None,
|
||||
checkpoint_callback=None,
|
||||
gradient_clip=0,
|
||||
cluster=None,
|
||||
process_position=0,
|
||||
@@ -44,20 +67,57 @@ class Trainer(TrainerIO):
|
||||
check_val_every_n_epoch=1,
|
||||
fast_dev_run=False,
|
||||
accumulate_grad_batches=1,
|
||||
enable_early_stop=True, max_nb_epochs=1000, min_nb_epochs=1,
|
||||
train_percent_check=1.0, val_percent_check=1.0, test_percent_check=1.0, val_check_interval=0.95,
|
||||
max_nb_epochs=1000, min_nb_epochs=1,
|
||||
train_percent_check=1.0, val_percent_check=1.0, test_percent_check=1.0,
|
||||
val_check_interval=0.95,
|
||||
log_save_interval=100, add_log_row_interval=10,
|
||||
lr_scheduler_milestones=None,
|
||||
distributed_backend='dp',
|
||||
use_amp=False,
|
||||
print_nan_grads=False,
|
||||
print_weights_summary=True,
|
||||
amp_level='O2',
|
||||
nb_sanity_val_steps=5):
|
||||
|
||||
"""
|
||||
|
||||
:param experiment: Test-tube experiment
|
||||
:param early_stop_callback: from pytorch_lightning import EarlyStopping
|
||||
:param checkpoint_callback: from pytorch_lightning import Checkpoint
|
||||
:param gradient_clip:
|
||||
:param cluster:
|
||||
:param process_position:
|
||||
:param current_gpu_name:
|
||||
:param nb_gpu_nodes:
|
||||
:param gpus:
|
||||
:param progress_bar:
|
||||
:param overfit_pct:
|
||||
:param track_grad_norm:
|
||||
:param check_val_every_n_epoch:
|
||||
:param fast_dev_run:
|
||||
:param accumulate_grad_batches:
|
||||
:param max_nb_epochs:
|
||||
:param min_nb_epochs:
|
||||
:param train_percent_check:
|
||||
:param val_percent_check:
|
||||
:param test_percent_check:
|
||||
:param val_check_interval:
|
||||
:param log_save_interval:
|
||||
:param add_log_row_interval:
|
||||
:param lr_scheduler_milestones:
|
||||
:param distributed_backend: 'np' to use DistributedParallel, 'ddp' to use DistributedDataParallel
|
||||
:param use_amp:
|
||||
:param print_nan_grads:
|
||||
:param print_weights_summary:
|
||||
:param amp_level:
|
||||
:param nb_sanity_val_steps:
|
||||
"""
|
||||
|
||||
# Transfer params
|
||||
self.nb_gpu_nodes = nb_gpu_nodes
|
||||
self.gradient_clip = gradient_clip
|
||||
self.check_val_every_n_epoch = check_val_every_n_epoch
|
||||
self.enable_early_stop = enable_early_stop
|
||||
self.enable_early_stop = early_stop_callback is not None
|
||||
self.track_grad_norm = track_grad_norm
|
||||
self.fast_dev_run = fast_dev_run
|
||||
self.on_gpu = gpus is not None and torch.cuda.is_available()
|
||||
@@ -67,8 +127,12 @@ class Trainer(TrainerIO):
|
||||
self.cluster = cluster
|
||||
self.process_position = process_position
|
||||
self.current_gpu_name = current_gpu_name
|
||||
self.print_weights_summary = print_weights_summary
|
||||
self.checkpoint_callback = checkpoint_callback
|
||||
self.checkpoint_callback.save_function = self.save_checkpoint
|
||||
|
||||
if self.checkpoint_callback is not None:
|
||||
self.checkpoint_callback.save_function = self.save_checkpoint
|
||||
|
||||
self.early_stop = early_stop_callback
|
||||
self.model = None
|
||||
self.max_nb_epochs = max_nb_epochs
|
||||
@@ -82,6 +146,8 @@ class Trainer(TrainerIO):
|
||||
self.print_nan_grads = print_nan_grads
|
||||
self.data_parallel_device_ids = None
|
||||
self.world_size = 1
|
||||
self.use_ddp = False
|
||||
self.use_dp = False
|
||||
|
||||
# gpus come in as a string.
|
||||
# if gpus = -1 then use all available devices
|
||||
@@ -95,8 +161,14 @@ class Trainer(TrainerIO):
|
||||
# set the correct cuda visible devices (using pci order)
|
||||
os.environ["CUDA_DEVICE_ORDER"] = "PCI_BUS_ID"
|
||||
os.environ["CUDA_VISIBLE_DEVICES"] = ','.join([str(x) for x in self.data_parallel_device_ids])
|
||||
print(f'VISIBLE GPUS: {os.environ["CUDA_VISIBLE_DEVICES"]}')
|
||||
|
||||
self.data_parallel = self.data_parallel_device_ids is not None and len(self.data_parallel_device_ids) > 1
|
||||
# make DP and DDP mutually exclusive
|
||||
# single GPU will also use DP with devices=[0]
|
||||
have_gpus = self.data_parallel_device_ids is not None and len(self.data_parallel_device_ids) > 0
|
||||
if have_gpus:
|
||||
self.use_dp = distributed_backend == 'dp'
|
||||
self.use_ddp = distributed_backend == 'ddp'
|
||||
|
||||
# process info
|
||||
self.proc_rank = 0
|
||||
@@ -137,6 +209,10 @@ class Trainer(TrainerIO):
|
||||
'''
|
||||
warnings.warn(msg)
|
||||
|
||||
@property
|
||||
def data_parallel(self):
|
||||
return self.use_dp or self.use_ddp
|
||||
|
||||
def __determine_data_use_amount(self, train_percent_check, val_percent_check, test_percent_check, overfit_pct):
|
||||
"""
|
||||
Use less data for debugging purposes
|
||||
@@ -212,9 +288,6 @@ class Trainer(TrainerIO):
|
||||
:param max_batches: Scalar
|
||||
:return:
|
||||
"""
|
||||
if self.proc_rank == 0:
|
||||
print('validating...')
|
||||
|
||||
# enable eval mode
|
||||
model.zero_grad()
|
||||
model.eval()
|
||||
@@ -238,8 +311,12 @@ class Trainer(TrainerIO):
|
||||
# -----------------
|
||||
# RUN VALIDATION STEP
|
||||
# -----------------
|
||||
if self.data_parallel:
|
||||
if self.use_ddp:
|
||||
output = model(data_batch, batch_i)
|
||||
elif self.use_dp:
|
||||
output = model(data_batch, batch_i)
|
||||
output = reduce_distributed_output(output, len(self.data_parallel_device_ids))
|
||||
|
||||
else:
|
||||
output = model.validation_step(data_batch, batch_i)
|
||||
|
||||
@@ -273,7 +350,7 @@ class Trainer(TrainerIO):
|
||||
self.test_dataloader = model.test_dataloader
|
||||
self.val_dataloader = model.val_dataloader
|
||||
|
||||
if self.data_parallel and type(self.tng_dataloader.sampler) is not DistributedSampler:
|
||||
if self.use_ddp and not isinstance(self.tng_dataloader.sampler, DistributedSampler):
|
||||
msg = '''
|
||||
when using multiple gpus and multiple nodes you must pass a DistributedSampler to DataLoader(sampler).
|
||||
|
||||
@@ -293,16 +370,34 @@ class Trainer(TrainerIO):
|
||||
# -----------------------------
|
||||
def fit(self, model):
|
||||
|
||||
# when using gpus, first thing we do is spawn a new process between each worker
|
||||
# multi-gpu and multi-nodes
|
||||
if self.data_parallel:
|
||||
# when using multi-node or DDP within a node start each module in a separate process
|
||||
if self.use_ddp:
|
||||
# must copy only the meta of the exp so it survives pickle/unpickle when going to new process
|
||||
self.experiment = self.experiment.get_meta_copy()
|
||||
mp.spawn(self.dp_train, nprocs=len(self.data_parallel_device_ids), args=(model, ))
|
||||
|
||||
# treat 1 gpu as a different case to avoid nccl bugs
|
||||
elif len(self.data_parallel_device_ids) == 1:
|
||||
self.single_gpu_train(model)
|
||||
# whenever we have the correct number of tasks, we let slurm manage processes
|
||||
# otherwise we launch the required number of processes
|
||||
nb_slurm_tasks = int(os.environ['SLURM_NTASKS'])
|
||||
nb_requested_gpus = len(self.data_parallel_device_ids)
|
||||
is_slurm_managing_tasks = nb_slurm_tasks == nb_requested_gpus
|
||||
if is_slurm_managing_tasks:
|
||||
task = int(os.environ['SLURM_LOCALID'])
|
||||
self.ddp_train(task, model)
|
||||
else:
|
||||
msg = f"""
|
||||
You requested {nb_requested_gpus} GPUs but launched {nb_slurm_tasks} slurm tasks.
|
||||
We will launch {nb_requested_gpus} processes for you.
|
||||
We recommend you let slurm manage the processes by setting: --ntasks-per-node={nb_requested_gpus}
|
||||
"""
|
||||
warnings.warn(msg)
|
||||
mp.spawn(self.ddp_train, nprocs=len(self.data_parallel_device_ids), args=(model, ))
|
||||
|
||||
# 1 gpu or dp option triggers training using DP module
|
||||
# easier to avoid NCCL issues
|
||||
elif self.use_dp:
|
||||
self.dp_train(model)
|
||||
|
||||
# ON CPU
|
||||
else:
|
||||
# CHOOSE OPTIMIZER
|
||||
# filter out the weights that were done on gpu so we can load on good old cpus
|
||||
@@ -318,14 +413,15 @@ class Trainer(TrainerIO):
|
||||
|
||||
self.__run_pretrain_routine(model)
|
||||
|
||||
def single_gpu_train(self, model):
|
||||
# torch.cuda.set_device(0)
|
||||
model.cuda(0)
|
||||
def dp_train(self, model):
|
||||
|
||||
# CHOOSE OPTIMIZER
|
||||
# filter out the weights that were done on gpu so we can load on good old cpus
|
||||
self.optimizers = model.configure_optimizers()
|
||||
|
||||
model.cuda(self.data_parallel_device_ids[0])
|
||||
model = LightningDataParallel(model, device_ids=self.data_parallel_device_ids)
|
||||
|
||||
# run through amp wrapper
|
||||
if self.use_amp:
|
||||
# An example
|
||||
@@ -336,7 +432,7 @@ class Trainer(TrainerIO):
|
||||
|
||||
self.__run_pretrain_routine(model)
|
||||
|
||||
def dp_train(self, gpu_nb, model):
|
||||
def ddp_train(self, gpu_nb, model):
|
||||
"""
|
||||
Entry point into a DP thread
|
||||
:param gpu_nb:
|
||||
@@ -443,7 +539,7 @@ class Trainer(TrainerIO):
|
||||
self.lr_schedulers.append(scheduler)
|
||||
|
||||
# print model summary
|
||||
if self.proc_rank == 0:
|
||||
if self.proc_rank == 0 and self.print_weights_summary:
|
||||
ref_model.summarize()
|
||||
|
||||
# give model convenience properties
|
||||
@@ -538,9 +634,11 @@ class Trainer(TrainerIO):
|
||||
if self.track_grad_norm > 0:
|
||||
model = self.__get_model()
|
||||
grad_norm_dic = model.grad_norm(self.track_grad_norm)
|
||||
|
||||
metrics.update(grad_norm_dic)
|
||||
|
||||
if self.__is_function_implemented('on_tng_metrics'):
|
||||
model.on_tng_metrics(metrics)
|
||||
|
||||
# log metrics
|
||||
scalar_metrics = self.__metrics_to_scalars(metrics, blacklist=self.__log_vals_blacklist())
|
||||
if self.proc_rank == 0:
|
||||
@@ -562,9 +660,9 @@ class Trainer(TrainerIO):
|
||||
model.on_epoch_end()
|
||||
|
||||
# early stopping
|
||||
if self.enable_early_stop:
|
||||
met_min_epochs = epoch_nb > self.min_nb_epochs
|
||||
if self.enable_early_stop and met_min_epochs:
|
||||
should_stop = self.early_stop_callback.on_epoch_end(epoch=epoch_nb, logs=self.__tng_tqdm_dic)
|
||||
met_min_epochs = epoch_nb > self.min_nb_epochs
|
||||
|
||||
# stop training
|
||||
stop = should_stop and met_min_epochs
|
||||
@@ -587,7 +685,7 @@ class Trainer(TrainerIO):
|
||||
|
||||
def __log_vals_blacklist(self):
|
||||
"""avoid logging some vals lightning uses to maintain state"""
|
||||
blacklist = {'batch_nb', 'v_nb', 'epoch', 'gpu'}
|
||||
blacklist = {'batch_nb', 'v_nb', 'gpu'}
|
||||
return blacklist
|
||||
|
||||
def __run_tng_batch(self, data_batch, batch_nb):
|
||||
@@ -596,8 +694,8 @@ class Trainer(TrainerIO):
|
||||
|
||||
# hook
|
||||
if self.__is_function_implemented('on_batch_start'):
|
||||
model = self.__get_model()
|
||||
response = model.on_batch_start(data_batch)
|
||||
model_ref = self.__get_model()
|
||||
response = model_ref.on_batch_start(data_batch)
|
||||
|
||||
if response == -1:
|
||||
return -1
|
||||
@@ -607,8 +705,11 @@ class Trainer(TrainerIO):
|
||||
|
||||
# forward pass
|
||||
# return a scalar value and a dic with tqdm metrics
|
||||
if self.data_parallel:
|
||||
if self.use_ddp:
|
||||
output = self.model(data_batch, batch_nb)
|
||||
elif self.use_dp:
|
||||
output = self.model(data_batch, batch_nb)
|
||||
output = reduce_distributed_output(output, len(self.data_parallel_device_ids))
|
||||
else:
|
||||
output = self.model.training_step(data_batch, batch_nb)
|
||||
|
||||
@@ -719,6 +820,6 @@ class Trainer(TrainerIO):
|
||||
self.prog_bar.set_postfix(**tqdm_metrics)
|
||||
|
||||
# model checkpointing
|
||||
if self.proc_rank == 0:
|
||||
if self.proc_rank == 0 and self.checkpoint_callback:
|
||||
print('save callback...')
|
||||
self.checkpoint_callback.on_epoch_end(epoch=self.current_epoch, logs=self.__tng_tqdm_dic)
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
from torch.nn import DataParallel
|
||||
from torch.nn.parallel import DistributedDataParallel
|
||||
import itertools
|
||||
from itertools import chain
|
||||
|
||||
import threading
|
||||
import torch
|
||||
@@ -42,6 +43,29 @@ class LightningDataParallel(DataParallel):
|
||||
Override the forward call in lightning so it goes to training and validation step respectively
|
||||
"""
|
||||
|
||||
def forward(self, *inputs, **kwargs):
|
||||
if not self.device_ids:
|
||||
return self.module(*inputs, **kwargs)
|
||||
|
||||
for t in chain(self.module.parameters(), self.module.buffers()):
|
||||
if t.device != self.src_device_obj:
|
||||
raise RuntimeError("module must have its parameters and buffers "
|
||||
"on device {} (device_ids[0]) but found one of "
|
||||
"them on device: {}".format(self.src_device_obj, t.device))
|
||||
|
||||
inputs, kwargs = self.scatter(inputs, kwargs, self.device_ids)
|
||||
if len(self.device_ids) == 1:
|
||||
# lightning
|
||||
if self.module.training:
|
||||
return self.module.training_step(*inputs[0], **kwargs[0])
|
||||
else:
|
||||
return self.module.validation_step(*inputs[0], **kwargs[0])
|
||||
|
||||
replicas = self.replicate(self.module, self.device_ids[:len(inputs)])
|
||||
outputs = self.parallel_apply(replicas, inputs, kwargs)
|
||||
return self.gather(outputs, self.output_device)
|
||||
|
||||
|
||||
def parallel_apply(self, replicas, inputs, kwargs):
|
||||
return parallel_apply(replicas, inputs, kwargs, self.device_ids[:len(replicas)])
|
||||
|
||||
|
||||
@@ -19,3 +19,6 @@ class ModelHooks(torch.nn.Module):
|
||||
def on_post_performance_check(self):
|
||||
pass
|
||||
|
||||
def on_tng_metrics(self, metrics):
|
||||
pass
|
||||
|
||||
|
||||
@@ -2,7 +2,7 @@ import torch
|
||||
import os
|
||||
import re
|
||||
import pdb
|
||||
from pytorch_lightning.pt_overrides.override_data_parallel import LightningDistributedDataParallel
|
||||
from pytorch_lightning.pt_overrides.override_data_parallel import LightningDistributedDataParallel, LightningDataParallel
|
||||
|
||||
class ModelIO(object):
|
||||
|
||||
@@ -66,7 +66,8 @@ class TrainerIO(object):
|
||||
checkpoint['optimizer_states'] = optimizer_states
|
||||
|
||||
# request what to save from the model
|
||||
model = self.model.module if type(self.model) is LightningDistributedDataParallel else self.model
|
||||
is_dp_module = type(self.model) is LightningDistributedDataParallel or type(self.model) is LightningDataParallel
|
||||
model = self.model.module if is_dp_module else self.model
|
||||
checkpoint_dict = model.get_save_dict()
|
||||
|
||||
# merge trainer and model saving items
|
||||
|
||||
@@ -78,7 +78,7 @@ class LightningModule(GradInformation, ModelIO, OptimizerConfig, ModelHooks):
|
||||
:param logs:
|
||||
:return:
|
||||
"""
|
||||
raise NotImplementedError
|
||||
return logs
|
||||
|
||||
def loss(self, *args, **kwargs):
|
||||
"""
|
||||
|
||||
@@ -7,7 +7,7 @@ from setuptools import setup, find_packages
|
||||
# http://blog.ionelmc.ro/2014/05/25/python-packaging/
|
||||
setup(
|
||||
name="pytorch-lightning",
|
||||
version='0.2.3',
|
||||
version='0.2.5.1',
|
||||
description="The Keras for ML researchers using PyTorch",
|
||||
author="William Falcon",
|
||||
author_email="waf2107@columbia.edu",
|
||||
@@ -19,8 +19,7 @@ setup(
|
||||
install_requires=[
|
||||
"torch>=1.1.0",
|
||||
"tqdm",
|
||||
"test-tube>=0.6.6",
|
||||
"tensorflow>=1.14.0"
|
||||
"test-tube>=0.6.7.1",
|
||||
],
|
||||
packages=find_packages(),
|
||||
long_description=open("README.md", encoding="utf-8").read(),
|
||||
|
||||
Reference in New Issue
Block a user