mirror of
https://github.com/wassname/pytorch-lightning.git
synced 2026-09-13 12:50:26 +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
|
return tqdm_dic
|
||||||
```
|
```
|
||||||
|
|
||||||
## TensorboardX
|
## Tensorboard
|
||||||
Lightning is fully integrated with tensorboardX.
|
Lightning is fully integrated with tensorboard.
|
||||||
|
|
||||||
<p align="center">
|
<p align="center">
|
||||||
<a href="https://williamfalcon.github.io/pytorch-lightning/">
|
<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)
|
trainer = Trainer(lr_scheduler_milestones=None)
|
||||||
|
|
||||||
# cut LR by 10 at 100, 200, and 300 epochs
|
# 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)
|
loss_val = self.loss(y, y_hat)
|
||||||
|
|
||||||
output = OrderedDict({
|
output = OrderedDict({
|
||||||
'loss': loss_val,
|
'loss': loss_val
|
||||||
'tqdm_metrics': {}
|
|
||||||
})
|
})
|
||||||
|
|
||||||
|
# can also return just a scalar instead of a dict (return loss_val)
|
||||||
return output
|
return output
|
||||||
|
|
||||||
def validation_step(self, data_batch, batch_i):
|
def validation_step(self, data_batch, batch_i):
|
||||||
@@ -107,8 +108,10 @@ class LightningTemplateModel(LightningModule):
|
|||||||
|
|
||||||
output = OrderedDict({
|
output = OrderedDict({
|
||||||
'val_loss': loss_val,
|
'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
|
return output
|
||||||
|
|
||||||
def validation_end(self, outputs):
|
def validation_end(self, outputs):
|
||||||
@@ -117,6 +120,10 @@ class LightningTemplateModel(LightningModule):
|
|||||||
:param outputs: list of individual outputs of each validation step
|
:param outputs: list of individual outputs of each validation step
|
||||||
:return:
|
: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_loss_mean = 0
|
||||||
val_acc_mean = 0
|
val_acc_mean = 0
|
||||||
for output in outputs:
|
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 traceback
|
||||||
import warnings
|
import warnings
|
||||||
import os
|
import os
|
||||||
|
import pdb
|
||||||
|
|
||||||
import torch
|
import torch
|
||||||
from torch.utils.data.distributed import DistributedSampler
|
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.memory import get_gpu_memory_map
|
||||||
from pytorch_lightning.root_module.model_saving import TrainerIO
|
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:
|
try:
|
||||||
@@ -27,11 +28,33 @@ except ModuleNotFoundError:
|
|||||||
APEX_AVAILABLE = False
|
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):
|
class Trainer(TrainerIO):
|
||||||
|
|
||||||
def __init__(self,
|
def __init__(self,
|
||||||
experiment,
|
experiment,
|
||||||
checkpoint_callback, early_stop_callback,
|
early_stop_callback=None,
|
||||||
|
checkpoint_callback=None,
|
||||||
gradient_clip=0,
|
gradient_clip=0,
|
||||||
cluster=None,
|
cluster=None,
|
||||||
process_position=0,
|
process_position=0,
|
||||||
@@ -44,20 +67,57 @@ class Trainer(TrainerIO):
|
|||||||
check_val_every_n_epoch=1,
|
check_val_every_n_epoch=1,
|
||||||
fast_dev_run=False,
|
fast_dev_run=False,
|
||||||
accumulate_grad_batches=1,
|
accumulate_grad_batches=1,
|
||||||
enable_early_stop=True, max_nb_epochs=1000, min_nb_epochs=1,
|
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,
|
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,
|
log_save_interval=100, add_log_row_interval=10,
|
||||||
lr_scheduler_milestones=None,
|
lr_scheduler_milestones=None,
|
||||||
|
distributed_backend='dp',
|
||||||
use_amp=False,
|
use_amp=False,
|
||||||
print_nan_grads=False,
|
print_nan_grads=False,
|
||||||
|
print_weights_summary=True,
|
||||||
amp_level='O2',
|
amp_level='O2',
|
||||||
nb_sanity_val_steps=5):
|
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
|
# Transfer params
|
||||||
self.nb_gpu_nodes = nb_gpu_nodes
|
self.nb_gpu_nodes = nb_gpu_nodes
|
||||||
self.gradient_clip = gradient_clip
|
self.gradient_clip = gradient_clip
|
||||||
self.check_val_every_n_epoch = check_val_every_n_epoch
|
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.track_grad_norm = track_grad_norm
|
||||||
self.fast_dev_run = fast_dev_run
|
self.fast_dev_run = fast_dev_run
|
||||||
self.on_gpu = gpus is not None and torch.cuda.is_available()
|
self.on_gpu = gpus is not None and torch.cuda.is_available()
|
||||||
@@ -67,8 +127,12 @@ class Trainer(TrainerIO):
|
|||||||
self.cluster = cluster
|
self.cluster = cluster
|
||||||
self.process_position = process_position
|
self.process_position = process_position
|
||||||
self.current_gpu_name = current_gpu_name
|
self.current_gpu_name = current_gpu_name
|
||||||
|
self.print_weights_summary = print_weights_summary
|
||||||
self.checkpoint_callback = checkpoint_callback
|
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.early_stop = early_stop_callback
|
||||||
self.model = None
|
self.model = None
|
||||||
self.max_nb_epochs = max_nb_epochs
|
self.max_nb_epochs = max_nb_epochs
|
||||||
@@ -82,6 +146,8 @@ class Trainer(TrainerIO):
|
|||||||
self.print_nan_grads = print_nan_grads
|
self.print_nan_grads = print_nan_grads
|
||||||
self.data_parallel_device_ids = None
|
self.data_parallel_device_ids = None
|
||||||
self.world_size = 1
|
self.world_size = 1
|
||||||
|
self.use_ddp = False
|
||||||
|
self.use_dp = False
|
||||||
|
|
||||||
# gpus come in as a string.
|
# gpus come in as a string.
|
||||||
# if gpus = -1 then use all available devices
|
# if gpus = -1 then use all available devices
|
||||||
@@ -95,8 +161,14 @@ class Trainer(TrainerIO):
|
|||||||
# set the correct cuda visible devices (using pci order)
|
# set the correct cuda visible devices (using pci order)
|
||||||
os.environ["CUDA_DEVICE_ORDER"] = "PCI_BUS_ID"
|
os.environ["CUDA_DEVICE_ORDER"] = "PCI_BUS_ID"
|
||||||
os.environ["CUDA_VISIBLE_DEVICES"] = ','.join([str(x) for x in self.data_parallel_device_ids])
|
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
|
# process info
|
||||||
self.proc_rank = 0
|
self.proc_rank = 0
|
||||||
@@ -137,6 +209,10 @@ class Trainer(TrainerIO):
|
|||||||
'''
|
'''
|
||||||
warnings.warn(msg)
|
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):
|
def __determine_data_use_amount(self, train_percent_check, val_percent_check, test_percent_check, overfit_pct):
|
||||||
"""
|
"""
|
||||||
Use less data for debugging purposes
|
Use less data for debugging purposes
|
||||||
@@ -212,9 +288,6 @@ class Trainer(TrainerIO):
|
|||||||
:param max_batches: Scalar
|
:param max_batches: Scalar
|
||||||
:return:
|
:return:
|
||||||
"""
|
"""
|
||||||
if self.proc_rank == 0:
|
|
||||||
print('validating...')
|
|
||||||
|
|
||||||
# enable eval mode
|
# enable eval mode
|
||||||
model.zero_grad()
|
model.zero_grad()
|
||||||
model.eval()
|
model.eval()
|
||||||
@@ -238,8 +311,12 @@ class Trainer(TrainerIO):
|
|||||||
# -----------------
|
# -----------------
|
||||||
# RUN VALIDATION STEP
|
# RUN VALIDATION STEP
|
||||||
# -----------------
|
# -----------------
|
||||||
if self.data_parallel:
|
if self.use_ddp:
|
||||||
output = model(data_batch, batch_i)
|
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:
|
else:
|
||||||
output = model.validation_step(data_batch, batch_i)
|
output = model.validation_step(data_batch, batch_i)
|
||||||
|
|
||||||
@@ -273,7 +350,7 @@ class Trainer(TrainerIO):
|
|||||||
self.test_dataloader = model.test_dataloader
|
self.test_dataloader = model.test_dataloader
|
||||||
self.val_dataloader = model.val_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 = '''
|
msg = '''
|
||||||
when using multiple gpus and multiple nodes you must pass a DistributedSampler to DataLoader(sampler).
|
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):
|
def fit(self, model):
|
||||||
|
|
||||||
# when using gpus, first thing we do is spawn a new process between each worker
|
# when using multi-node or DDP within a node start each module in a separate process
|
||||||
# multi-gpu and multi-nodes
|
if self.use_ddp:
|
||||||
if self.data_parallel:
|
# 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()
|
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
|
# whenever we have the correct number of tasks, we let slurm manage processes
|
||||||
elif len(self.data_parallel_device_ids) == 1:
|
# otherwise we launch the required number of processes
|
||||||
self.single_gpu_train(model)
|
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:
|
else:
|
||||||
# CHOOSE OPTIMIZER
|
# CHOOSE OPTIMIZER
|
||||||
# filter out the weights that were done on gpu so we can load on good old cpus
|
# 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)
|
self.__run_pretrain_routine(model)
|
||||||
|
|
||||||
def single_gpu_train(self, model):
|
def dp_train(self, model):
|
||||||
# torch.cuda.set_device(0)
|
|
||||||
model.cuda(0)
|
|
||||||
|
|
||||||
# CHOOSE OPTIMIZER
|
# CHOOSE OPTIMIZER
|
||||||
# filter out the weights that were done on gpu so we can load on good old cpus
|
# filter out the weights that were done on gpu so we can load on good old cpus
|
||||||
self.optimizers = model.configure_optimizers()
|
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
|
# run through amp wrapper
|
||||||
if self.use_amp:
|
if self.use_amp:
|
||||||
# An example
|
# An example
|
||||||
@@ -336,7 +432,7 @@ class Trainer(TrainerIO):
|
|||||||
|
|
||||||
self.__run_pretrain_routine(model)
|
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
|
Entry point into a DP thread
|
||||||
:param gpu_nb:
|
:param gpu_nb:
|
||||||
@@ -443,7 +539,7 @@ class Trainer(TrainerIO):
|
|||||||
self.lr_schedulers.append(scheduler)
|
self.lr_schedulers.append(scheduler)
|
||||||
|
|
||||||
# print model summary
|
# print model summary
|
||||||
if self.proc_rank == 0:
|
if self.proc_rank == 0 and self.print_weights_summary:
|
||||||
ref_model.summarize()
|
ref_model.summarize()
|
||||||
|
|
||||||
# give model convenience properties
|
# give model convenience properties
|
||||||
@@ -538,9 +634,11 @@ class Trainer(TrainerIO):
|
|||||||
if self.track_grad_norm > 0:
|
if self.track_grad_norm > 0:
|
||||||
model = self.__get_model()
|
model = self.__get_model()
|
||||||
grad_norm_dic = model.grad_norm(self.track_grad_norm)
|
grad_norm_dic = model.grad_norm(self.track_grad_norm)
|
||||||
|
|
||||||
metrics.update(grad_norm_dic)
|
metrics.update(grad_norm_dic)
|
||||||
|
|
||||||
|
if self.__is_function_implemented('on_tng_metrics'):
|
||||||
|
model.on_tng_metrics(metrics)
|
||||||
|
|
||||||
# log metrics
|
# log metrics
|
||||||
scalar_metrics = self.__metrics_to_scalars(metrics, blacklist=self.__log_vals_blacklist())
|
scalar_metrics = self.__metrics_to_scalars(metrics, blacklist=self.__log_vals_blacklist())
|
||||||
if self.proc_rank == 0:
|
if self.proc_rank == 0:
|
||||||
@@ -562,9 +660,9 @@ class Trainer(TrainerIO):
|
|||||||
model.on_epoch_end()
|
model.on_epoch_end()
|
||||||
|
|
||||||
# early stopping
|
# 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)
|
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 training
|
||||||
stop = should_stop and met_min_epochs
|
stop = should_stop and met_min_epochs
|
||||||
@@ -587,7 +685,7 @@ class Trainer(TrainerIO):
|
|||||||
|
|
||||||
def __log_vals_blacklist(self):
|
def __log_vals_blacklist(self):
|
||||||
"""avoid logging some vals lightning uses to maintain state"""
|
"""avoid logging some vals lightning uses to maintain state"""
|
||||||
blacklist = {'batch_nb', 'v_nb', 'epoch', 'gpu'}
|
blacklist = {'batch_nb', 'v_nb', 'gpu'}
|
||||||
return blacklist
|
return blacklist
|
||||||
|
|
||||||
def __run_tng_batch(self, data_batch, batch_nb):
|
def __run_tng_batch(self, data_batch, batch_nb):
|
||||||
@@ -596,8 +694,8 @@ class Trainer(TrainerIO):
|
|||||||
|
|
||||||
# hook
|
# hook
|
||||||
if self.__is_function_implemented('on_batch_start'):
|
if self.__is_function_implemented('on_batch_start'):
|
||||||
model = self.__get_model()
|
model_ref = self.__get_model()
|
||||||
response = model.on_batch_start(data_batch)
|
response = model_ref.on_batch_start(data_batch)
|
||||||
|
|
||||||
if response == -1:
|
if response == -1:
|
||||||
return -1
|
return -1
|
||||||
@@ -607,8 +705,11 @@ class Trainer(TrainerIO):
|
|||||||
|
|
||||||
# forward pass
|
# forward pass
|
||||||
# return a scalar value and a dic with tqdm metrics
|
# 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)
|
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:
|
else:
|
||||||
output = self.model.training_step(data_batch, batch_nb)
|
output = self.model.training_step(data_batch, batch_nb)
|
||||||
|
|
||||||
@@ -719,6 +820,6 @@ class Trainer(TrainerIO):
|
|||||||
self.prog_bar.set_postfix(**tqdm_metrics)
|
self.prog_bar.set_postfix(**tqdm_metrics)
|
||||||
|
|
||||||
# model checkpointing
|
# model checkpointing
|
||||||
if self.proc_rank == 0:
|
if self.proc_rank == 0 and self.checkpoint_callback:
|
||||||
print('save callback...')
|
print('save callback...')
|
||||||
self.checkpoint_callback.on_epoch_end(epoch=self.current_epoch, logs=self.__tng_tqdm_dic)
|
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 import DataParallel
|
||||||
from torch.nn.parallel import DistributedDataParallel
|
from torch.nn.parallel import DistributedDataParallel
|
||||||
import itertools
|
import itertools
|
||||||
|
from itertools import chain
|
||||||
|
|
||||||
import threading
|
import threading
|
||||||
import torch
|
import torch
|
||||||
@@ -42,6 +43,29 @@ class LightningDataParallel(DataParallel):
|
|||||||
Override the forward call in lightning so it goes to training and validation step respectively
|
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):
|
def parallel_apply(self, replicas, inputs, kwargs):
|
||||||
return parallel_apply(replicas, inputs, kwargs, self.device_ids[:len(replicas)])
|
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):
|
def on_post_performance_check(self):
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
def on_tng_metrics(self, metrics):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|||||||
@@ -2,7 +2,7 @@ import torch
|
|||||||
import os
|
import os
|
||||||
import re
|
import re
|
||||||
import pdb
|
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):
|
class ModelIO(object):
|
||||||
|
|
||||||
@@ -66,7 +66,8 @@ class TrainerIO(object):
|
|||||||
checkpoint['optimizer_states'] = optimizer_states
|
checkpoint['optimizer_states'] = optimizer_states
|
||||||
|
|
||||||
# request what to save from the model
|
# 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()
|
checkpoint_dict = model.get_save_dict()
|
||||||
|
|
||||||
# merge trainer and model saving items
|
# merge trainer and model saving items
|
||||||
|
|||||||
@@ -78,7 +78,7 @@ class LightningModule(GradInformation, ModelIO, OptimizerConfig, ModelHooks):
|
|||||||
:param logs:
|
:param logs:
|
||||||
:return:
|
:return:
|
||||||
"""
|
"""
|
||||||
raise NotImplementedError
|
return logs
|
||||||
|
|
||||||
def loss(self, *args, **kwargs):
|
def loss(self, *args, **kwargs):
|
||||||
"""
|
"""
|
||||||
|
|||||||
@@ -7,7 +7,7 @@ from setuptools import setup, find_packages
|
|||||||
# http://blog.ionelmc.ro/2014/05/25/python-packaging/
|
# http://blog.ionelmc.ro/2014/05/25/python-packaging/
|
||||||
setup(
|
setup(
|
||||||
name="pytorch-lightning",
|
name="pytorch-lightning",
|
||||||
version='0.2.3',
|
version='0.2.5.1',
|
||||||
description="The Keras for ML researchers using PyTorch",
|
description="The Keras for ML researchers using PyTorch",
|
||||||
author="William Falcon",
|
author="William Falcon",
|
||||||
author_email="waf2107@columbia.edu",
|
author_email="waf2107@columbia.edu",
|
||||||
@@ -19,8 +19,7 @@ setup(
|
|||||||
install_requires=[
|
install_requires=[
|
||||||
"torch>=1.1.0",
|
"torch>=1.1.0",
|
||||||
"tqdm",
|
"tqdm",
|
||||||
"test-tube>=0.6.6",
|
"test-tube>=0.6.7.1",
|
||||||
"tensorflow>=1.14.0"
|
|
||||||
],
|
],
|
||||||
packages=find_packages(),
|
packages=find_packages(),
|
||||||
long_description=open("README.md", encoding="utf-8").read(),
|
long_description=open("README.md", encoding="utf-8").read(),
|
||||||
|
|||||||
Reference in New Issue
Block a user