|
| 1 | +import argparse |
| 2 | +import json |
| 3 | +import logging |
| 4 | +import os |
| 5 | +import sys |
| 6 | +import torch |
| 7 | +import torch.distributed as dist |
| 8 | +import torch.nn as nn |
| 9 | +import torch.nn.functional as F |
| 10 | +import torch.optim as optim |
| 11 | +import torch.utils.data |
| 12 | +import torch.utils.data.distributed |
| 13 | +from torchvision import datasets, transforms |
| 14 | + |
| 15 | +logger = logging.getLogger(__name__) |
| 16 | +logger.setLevel(logging.DEBUG) |
| 17 | +logger.addHandler(logging.StreamHandler(sys.stdout)) |
| 18 | + |
| 19 | + |
| 20 | +class Net(nn.Module): |
| 21 | + # Based on https://github.com/pytorch/examples/blob/master/mnist/main.py |
| 22 | + def __init__(self): |
| 23 | + logger.info("Create neural network module") |
| 24 | + |
| 25 | + super(Net, self).__init__() |
| 26 | + self.conv1 = nn.Conv2d(1, 10, kernel_size=5) |
| 27 | + self.conv2 = nn.Conv2d(10, 20, kernel_size=5) |
| 28 | + self.conv2_drop = nn.Dropout2d() |
| 29 | + self.fc1 = nn.Linear(320, 50) |
| 30 | + self.fc2 = nn.Linear(50, 10) |
| 31 | + |
| 32 | + def forward(self, x): |
| 33 | + x = F.relu(F.max_pool2d(self.conv1(x), 2)) |
| 34 | + x = F.relu(F.max_pool2d(self.conv2_drop(self.conv2(x)), 2)) |
| 35 | + x = x.view(-1, 320) |
| 36 | + x = F.relu(self.fc1(x)) |
| 37 | + x = F.dropout(x, training=self.training) |
| 38 | + x = self.fc2(x) |
| 39 | + return F.log_softmax(x, dim=1) |
| 40 | + |
| 41 | + |
| 42 | +def _get_train_data_loader(training_dir, is_distributed, batch_size, **kwargs): |
| 43 | + logger.info("Get train data loader") |
| 44 | + dataset = datasets.MNIST( |
| 45 | + training_dir, |
| 46 | + train=True, |
| 47 | + transform=transforms.Compose( |
| 48 | + [transforms.ToTensor(), transforms.Normalize((0.1307,), (0.3081,))] |
| 49 | + ), |
| 50 | + download=False, # True sets a dependency on an external site for our canaries. |
| 51 | + ) |
| 52 | + train_sampler = ( |
| 53 | + torch.utils.data.distributed.DistributedSampler(dataset) if is_distributed else None |
| 54 | + ) |
| 55 | + train_loader = torch.utils.data.DataLoader( |
| 56 | + dataset, |
| 57 | + batch_size=batch_size, |
| 58 | + shuffle=train_sampler is None, |
| 59 | + sampler=train_sampler, |
| 60 | + **kwargs |
| 61 | + ) |
| 62 | + return train_sampler, train_loader |
| 63 | + |
| 64 | + |
| 65 | +def _get_test_data_loader(training_dir, **kwargs): |
| 66 | + logger.info("Get test data loader") |
| 67 | + return torch.utils.data.DataLoader( |
| 68 | + datasets.MNIST( |
| 69 | + training_dir, |
| 70 | + train=False, |
| 71 | + transform=transforms.Compose( |
| 72 | + [transforms.ToTensor(), transforms.Normalize((0.1307,), (0.3081,))] |
| 73 | + ), |
| 74 | + download=False, # True sets a dependency on an external site for our canaries. |
| 75 | + ), |
| 76 | + batch_size=1000, |
| 77 | + shuffle=True, |
| 78 | + **kwargs |
| 79 | + ) |
| 80 | + |
| 81 | + |
| 82 | +def _average_gradients(model): |
| 83 | + # Gradient averaging. |
| 84 | + size = float(dist.get_world_size()) |
| 85 | + for param in model.parameters(): |
| 86 | + dist.all_reduce(param.grad.data, op=dist.reduce_op.SUM, group=0) |
| 87 | + param.grad.data /= size |
| 88 | + |
| 89 | + |
| 90 | +def train(args): |
| 91 | + world_size = len(args.hosts) |
| 92 | + is_distributed = world_size > 1 |
| 93 | + logger.debug("Number of hosts {}. Distributed training - {}".format(world_size, is_distributed)) |
| 94 | + use_cuda = args.num_gpus > 0 |
| 95 | + logger.debug("Number of gpus available - {}".format(args.num_gpus)) |
| 96 | + kwargs = {"num_workers": 1, "pin_memory": True} if use_cuda else {} |
| 97 | + device = torch.device("cuda" if use_cuda else "cpu") |
| 98 | + |
| 99 | + if is_distributed: |
| 100 | + # Initialize the distributed environment. |
| 101 | + backend = "gloo" |
| 102 | + os.environ["WORLD_SIZE"] = str(world_size) |
| 103 | + host_rank = args.hosts.index(args.current_host) |
| 104 | + dist.init_process_group(backend=backend, rank=host_rank, world_size=world_size) |
| 105 | + logger.info( |
| 106 | + "Initialized the distributed environment: '{}' backend on {} nodes. ".format( |
| 107 | + backend, dist.get_world_size() |
| 108 | + ) |
| 109 | + + "Current host rank is {}. Is cuda available: {}. Number of gpus: {}".format( |
| 110 | + dist.get_rank(), torch.cuda.is_available(), args.num_gpus |
| 111 | + ) |
| 112 | + ) |
| 113 | + |
| 114 | + # set the seed for generating random numbers |
| 115 | + seed = 1 |
| 116 | + torch.manual_seed(seed) |
| 117 | + if use_cuda: |
| 118 | + torch.cuda.manual_seed(seed) |
| 119 | + |
| 120 | + train_sampler, train_loader = _get_train_data_loader( |
| 121 | + args.data_dir, is_distributed, args.batch_size, **kwargs |
| 122 | + ) |
| 123 | + test_loader = _get_test_data_loader(args.data_dir, **kwargs) |
| 124 | + |
| 125 | + logger.debug( |
| 126 | + "Processes {}/{} ({:.0f}%) of train data".format( |
| 127 | + len(train_loader.sampler), |
| 128 | + len(train_loader.dataset), |
| 129 | + 100.0 * len(train_loader.sampler) / len(train_loader.dataset), |
| 130 | + ) |
| 131 | + ) |
| 132 | + |
| 133 | + logger.debug( |
| 134 | + "Processes {}/{} ({:.0f}%) of test data".format( |
| 135 | + len(test_loader.sampler), |
| 136 | + len(test_loader.dataset), |
| 137 | + 100.0 * len(test_loader.sampler) / len(test_loader.dataset), |
| 138 | + ) |
| 139 | + ) |
| 140 | + |
| 141 | + model = Net().to(device) |
| 142 | + if is_distributed and use_cuda: |
| 143 | + # multi-machine multi-gpu case |
| 144 | + logger.debug("Multi-machine multi-gpu: using DistributedDataParallel.") |
| 145 | + model = torch.nn.parallel.DistributedDataParallel(model) |
| 146 | + elif use_cuda: |
| 147 | + # single-machine multi-gpu case |
| 148 | + logger.debug("Single-machine multi-gpu: using DataParallel().cuda().") |
| 149 | + model = torch.nn.DataParallel(model) |
| 150 | + else: |
| 151 | + # single-machine or multi-machine cpu case |
| 152 | + logger.debug("Single-machine/multi-machine cpu: using DataParallel.") |
| 153 | + model = torch.nn.DataParallel(model) |
| 154 | + |
| 155 | + optimizer = optim.SGD(model.parameters(), lr=0.1, momentum=0.5) |
| 156 | + |
| 157 | + log_interval = 100 |
| 158 | + for epoch in range(1, args.epochs + 1): |
| 159 | + if is_distributed: |
| 160 | + train_sampler.set_epoch(epoch) |
| 161 | + model.train() |
| 162 | + for batch_idx, (data, target) in enumerate(train_loader, 1): |
| 163 | + data, target = data.to(device), target.to(device) |
| 164 | + optimizer.zero_grad() |
| 165 | + output = model(data) |
| 166 | + loss = F.nll_loss(output, target) |
| 167 | + loss.backward() |
| 168 | + if is_distributed and not use_cuda: |
| 169 | + # average gradients manually for multi-machine cpu case only |
| 170 | + _average_gradients(model) |
| 171 | + optimizer.step() |
| 172 | + if batch_idx % log_interval == 0: |
| 173 | + logger.debug( |
| 174 | + "Train Epoch: {} [{}/{} ({:.0f}%)] Loss: {:.6f}".format( |
| 175 | + epoch, |
| 176 | + batch_idx * len(data), |
| 177 | + len(train_loader.sampler), |
| 178 | + 100.0 * batch_idx / len(train_loader), |
| 179 | + loss.item(), |
| 180 | + ) |
| 181 | + ) |
| 182 | + accuracy = test(model, test_loader, device) |
| 183 | + save_model(model, args.model_dir) |
| 184 | + |
| 185 | + logger.debug("Overall test accuracy: {};".format(accuracy)) |
| 186 | + |
| 187 | + |
| 188 | +def test(model, test_loader, device): |
| 189 | + model.eval() |
| 190 | + test_loss = 0 |
| 191 | + correct = 0 |
| 192 | + with torch.no_grad(): |
| 193 | + for data, target in test_loader: |
| 194 | + data, target = data.to(device), target.to(device) |
| 195 | + output = model(data) |
| 196 | + test_loss += F.nll_loss(output, target, size_average=False).item() # sum up batch loss |
| 197 | + pred = output.max(1, keepdim=True)[1] # get the index of the max log-probability |
| 198 | + correct += pred.eq(target.view_as(pred)).sum().item() |
| 199 | + |
| 200 | + test_loss /= len(test_loader.dataset) |
| 201 | + accuracy = 100.0 * correct / len(test_loader.dataset) |
| 202 | + |
| 203 | + logger.debug( |
| 204 | + "Test set: Average loss: {:.4f}, Accuracy: {}/{} ({:.0f}%)\n".format( |
| 205 | + test_loss, correct, len(test_loader.dataset), accuracy |
| 206 | + ) |
| 207 | + ) |
| 208 | + |
| 209 | + return accuracy |
| 210 | + |
| 211 | + |
| 212 | +def model_fn(model_dir): |
| 213 | + model = torch.nn.DataParallel(Net()) |
| 214 | + with open(os.path.join(model_dir, "model.pth"), "rb") as f: |
| 215 | + model.load_state_dict(torch.load(f)) |
| 216 | + return model |
| 217 | + |
| 218 | + |
| 219 | +def save_model(model, model_dir): |
| 220 | + logger.info("Saving the model.") |
| 221 | + path = os.path.join(model_dir, "model.pth") |
| 222 | + # recommended way from http://pytorch.org/docs/master/notes/serialization.html |
| 223 | + torch.save(model.state_dict(), path) |
| 224 | + |
| 225 | + |
| 226 | +if __name__ == "__main__": |
| 227 | + parser = argparse.ArgumentParser() |
| 228 | + parser.add_argument("--epochs", type=int, default=1, metavar="N") |
| 229 | + parser.add_argument("--batch-size", type=int, default=64, metavar="N") |
| 230 | + |
| 231 | + # Container environment |
| 232 | + parser.add_argument("--hosts", type=list, default=json.loads(os.environ["SM_HOSTS"])) |
| 233 | + parser.add_argument("--current-host", type=str, default=os.environ["SM_CURRENT_HOST"]) |
| 234 | + parser.add_argument("--model-dir", type=str, default=os.environ["SM_MODEL_DIR"]) |
| 235 | + parser.add_argument("--data-dir", type=str, default=os.environ["SM_CHANNEL_TRAINING"]) |
| 236 | + parser.add_argument("--num-gpus", type=int, default=os.environ["SM_NUM_GPUS"]) |
| 237 | + parser.add_argument("--num-cpus", type=int, default=os.environ["SM_NUM_CPUS"]) |
| 238 | + |
| 239 | + train(parser.parse_args()) |
0 commit comments