|
| 1 | +# |
| 2 | +# Licensed to the Apache Software Foundation (ASF) under one |
| 3 | +# or more contributor license agreements. See the NOTICE file |
| 4 | +# distributed with this work for additional information |
| 5 | +# regarding copyright ownership. The ASF licenses this file |
| 6 | +# to you under the Apache License, Version 2.0 (the |
| 7 | +# "License"); you may not use this file except in compliance |
| 8 | +# with the License. You may obtain a copy of the License at |
| 9 | +# |
| 10 | +# http://www.apache.org/licenses/LICENSE-2.0 |
| 11 | +# |
| 12 | +# Unless required by applicable law or agreed to in writing, |
| 13 | +# software distributed under the License is distributed on an |
| 14 | +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| 15 | +# KIND, either express or implied. See the License for the |
| 16 | +# specific language governing permissions and limitations |
| 17 | +# under the License. |
| 18 | +# |
| 19 | + |
| 20 | +from singa import singa_wrap as singa |
| 21 | +from singa import autograd |
| 22 | +from singa import layer |
| 23 | +from singa import tensor |
| 24 | +from singa import device |
| 25 | +from singa import opt |
| 26 | +import numpy as np |
| 27 | +import os |
| 28 | +import sys |
| 29 | +import gzip |
| 30 | +import codecs |
| 31 | +import time |
| 32 | + |
| 33 | + |
| 34 | +class CNN: |
| 35 | + |
| 36 | + def __init__(self): |
| 37 | + self.conv1 = layer.Conv2d(1, 20, 5, padding=0) |
| 38 | + self.conv2 = layer.Conv2d(20, 50, 5, padding=0) |
| 39 | + self.linear1 = layer.Linear(4 * 4 * 50, 500) |
| 40 | + self.linear2 = layer.Linear(500, 10) |
| 41 | + self.pooling1 = layer.MaxPool2d(2, 2, padding=0) |
| 42 | + self.pooling2 = layer.MaxPool2d(2, 2, padding=0) |
| 43 | + self.relu1 = layer.ReLU() |
| 44 | + self.relu2 = layer.ReLU() |
| 45 | + self.relu3 = layer.ReLU() |
| 46 | + self.flatten = layer.Flatten() |
| 47 | + |
| 48 | + def forward(self, x): |
| 49 | + y = self.conv1(x) |
| 50 | + y = self.relu1(y) |
| 51 | + y = self.pooling1(y) |
| 52 | + y = self.conv2(y) |
| 53 | + y = self.relu2(y) |
| 54 | + y = self.pooling2(y) |
| 55 | + y = self.flatten(y) |
| 56 | + y = self.linear1(y) |
| 57 | + y = self.relu3(y) |
| 58 | + y = self.linear2(y) |
| 59 | + return y |
| 60 | + |
| 61 | + |
| 62 | +def check_dataset_exist(dirpath): |
| 63 | + if not os.path.exists(dirpath): |
| 64 | + print( |
| 65 | + 'The MNIST dataset does not exist. Please download the mnist dataset using download_mnist.py (e.g. python3 download_mnist.py)' |
| 66 | + ) |
| 67 | + sys.exit(0) |
| 68 | + return dirpath |
| 69 | + |
| 70 | + |
| 71 | +def load_dataset(): |
| 72 | + train_x_path = '/tmp/train-images-idx3-ubyte.gz' |
| 73 | + train_y_path = '/tmp/train-labels-idx1-ubyte.gz' |
| 74 | + valid_x_path = '/tmp/t10k-images-idx3-ubyte.gz' |
| 75 | + valid_y_path = '/tmp/t10k-labels-idx1-ubyte.gz' |
| 76 | + |
| 77 | + train_x = read_image_file(check_dataset_exist(train_x_path)).astype( |
| 78 | + np.float32) |
| 79 | + train_y = read_label_file(check_dataset_exist(train_y_path)).astype( |
| 80 | + np.float32) |
| 81 | + valid_x = read_image_file(check_dataset_exist(valid_x_path)).astype( |
| 82 | + np.float32) |
| 83 | + valid_y = read_label_file(check_dataset_exist(valid_y_path)).astype( |
| 84 | + np.float32) |
| 85 | + return train_x, train_y, valid_x, valid_y |
| 86 | + |
| 87 | + |
| 88 | +def read_label_file(path): |
| 89 | + with gzip.open(path, 'rb') as f: |
| 90 | + data = f.read() |
| 91 | + assert get_int(data[:4]) == 2049 |
| 92 | + length = get_int(data[4:8]) |
| 93 | + parsed = np.frombuffer(data, dtype=np.uint8, offset=8).reshape((length)) |
| 94 | + return parsed |
| 95 | + |
| 96 | + |
| 97 | +def get_int(b): |
| 98 | + return int(codecs.encode(b, 'hex'), 16) |
| 99 | + |
| 100 | + |
| 101 | +def read_image_file(path): |
| 102 | + with gzip.open(path, 'rb') as f: |
| 103 | + data = f.read() |
| 104 | + assert get_int(data[:4]) == 2051 |
| 105 | + length = get_int(data[4:8]) |
| 106 | + num_rows = get_int(data[8:12]) |
| 107 | + num_cols = get_int(data[12:16]) |
| 108 | + parsed = np.frombuffer(data, dtype=np.uint8, offset=16).reshape( |
| 109 | + (length, 1, num_rows, num_cols)) |
| 110 | + return parsed |
| 111 | + |
| 112 | + |
| 113 | +def to_categorical(y, num_classes): |
| 114 | + y = np.array(y, dtype="int") |
| 115 | + n = y.shape[0] |
| 116 | + categorical = np.zeros((n, num_classes)) |
| 117 | + categorical[np.arange(n), y] = 1 |
| 118 | + categorical = categorical.astype(np.float32) |
| 119 | + return categorical |
| 120 | + |
| 121 | + |
| 122 | +def accuracy(pred, target): |
| 123 | + y = np.argmax(pred, axis=1) |
| 124 | + t = np.argmax(target, axis=1) |
| 125 | + a = y == t |
| 126 | + return np.array(a, "int").sum() |
| 127 | + |
| 128 | + |
| 129 | +# Function to all reduce NUMPY accuracy and loss from multiple devices |
| 130 | +def reduce_variable(variable, dist_opt, reducer): |
| 131 | + reducer.copy_from_numpy(variable) |
| 132 | + dist_opt.all_reduce(reducer.data) |
| 133 | + dist_opt.wait() |
| 134 | + output = tensor.to_numpy(reducer) |
| 135 | + return output |
| 136 | + |
| 137 | + |
| 138 | +# Function to sychronize SINGA TENSOR initial model parameters |
| 139 | +def synchronize(tensor, dist_opt): |
| 140 | + dist_opt.all_reduce(tensor.data) |
| 141 | + dist_opt.wait() |
| 142 | + tensor /= dist_opt.world_size |
| 143 | + |
| 144 | + |
| 145 | +# Data augmentation |
| 146 | +def augmentation(x, batch_size): |
| 147 | + xpad = np.pad(x, [[0, 0], [0, 0], [4, 4], [4, 4]], 'symmetric') |
| 148 | + for data_num in range(0, batch_size): |
| 149 | + offset = np.random.randint(8, size=2) |
| 150 | + x[data_num, :, :, :] = xpad[data_num, :, offset[0]:offset[0] + 28, |
| 151 | + offset[1]:offset[1] + 28] |
| 152 | + if_flip = np.random.randint(2) |
| 153 | + if (if_flip): |
| 154 | + x[data_num, :, :, :] = x[data_num, :, :, ::-1] |
| 155 | + return x |
| 156 | + |
| 157 | + |
| 158 | +# Data partition |
| 159 | +def data_partition(dataset_x, dataset_y, global_rank, world_size): |
| 160 | + data_per_rank = dataset_x.shape[0] // world_size |
| 161 | + idx_start = global_rank * data_per_rank |
| 162 | + idx_end = (global_rank + 1) * data_per_rank |
| 163 | + return dataset_x[idx_start:idx_end], dataset_y[idx_start:idx_end] |
| 164 | + |
| 165 | + |
| 166 | +def train_mnist_cnn(DIST=False, |
| 167 | + local_rank=None, |
| 168 | + world_size=None, |
| 169 | + nccl_id=None, |
| 170 | + spars=0, |
| 171 | + topK=False, |
| 172 | + corr=True): |
| 173 | + |
| 174 | + # Define the hypermeters for the mnist_cnn |
| 175 | + max_epoch = 10 |
| 176 | + batch_size = 64 |
| 177 | + sgd = opt.SGD(lr=0.005, momentum=0.9, weight_decay=1e-5) |
| 178 | + |
| 179 | + # Prepare training and valadiation data |
| 180 | + train_x, train_y, test_x, test_y = load_dataset() |
| 181 | + IMG_SIZE = 28 |
| 182 | + num_classes = 10 |
| 183 | + train_y = to_categorical(train_y, num_classes) |
| 184 | + test_y = to_categorical(test_y, num_classes) |
| 185 | + |
| 186 | + # Normalization |
| 187 | + train_x = train_x / 255 |
| 188 | + test_x = test_x / 255 |
| 189 | + |
| 190 | + if DIST: |
| 191 | + # For distributed GPU training |
| 192 | + sgd = opt.DistOpt(sgd, |
| 193 | + nccl_id=nccl_id, |
| 194 | + local_rank=local_rank, |
| 195 | + world_size=world_size) |
| 196 | + dev = device.create_cuda_gpu_on(sgd.local_rank) |
| 197 | + |
| 198 | + # Dataset partition for distributed training |
| 199 | + train_x, train_y = data_partition(train_x, train_y, sgd.global_rank, |
| 200 | + sgd.world_size) |
| 201 | + test_x, test_y = data_partition(test_x, test_y, sgd.global_rank, |
| 202 | + sgd.world_size) |
| 203 | + world_size = sgd.world_size |
| 204 | + else: |
| 205 | + # For single GPU |
| 206 | + dev = device.create_cuda_gpu() |
| 207 | + world_size = 1 |
| 208 | + |
| 209 | + # Create model |
| 210 | + model = CNN() |
| 211 | + |
| 212 | + tx = tensor.Tensor((batch_size, 1, IMG_SIZE, IMG_SIZE), dev, tensor.float32) |
| 213 | + ty = tensor.Tensor((batch_size, num_classes), dev, tensor.int32) |
| 214 | + num_train_batch = train_x.shape[0] // batch_size |
| 215 | + num_test_batch = test_x.shape[0] // batch_size |
| 216 | + idx = np.arange(train_x.shape[0], dtype=np.int32) |
| 217 | + |
| 218 | + if DIST: |
| 219 | + #Sychronize the initial parameters |
| 220 | + autograd.training = True |
| 221 | + x = np.random.randn(batch_size, 1, IMG_SIZE, |
| 222 | + IMG_SIZE).astype(np.float32) |
| 223 | + y = np.zeros(shape=(batch_size, num_classes), dtype=np.int32) |
| 224 | + tx.copy_from_numpy(x) |
| 225 | + ty.copy_from_numpy(y) |
| 226 | + out = model.forward(tx) |
| 227 | + loss = autograd.softmax_cross_entropy(out, ty) |
| 228 | + for p, g in autograd.backward(loss): |
| 229 | + synchronize(p, sgd) |
| 230 | + |
| 231 | + # Training and evaulation loop |
| 232 | + for epoch in range(max_epoch): |
| 233 | + start_time = time.time() |
| 234 | + np.random.shuffle(idx) |
| 235 | + |
| 236 | + if ((DIST == False) or (sgd.global_rank == 0)): |
| 237 | + print('Starting Epoch %d:' % (epoch)) |
| 238 | + |
| 239 | + # Training phase |
| 240 | + autograd.training = True |
| 241 | + train_correct = np.zeros(shape=[1], dtype=np.float32) |
| 242 | + test_correct = np.zeros(shape=[1], dtype=np.float32) |
| 243 | + train_loss = np.zeros(shape=[1], dtype=np.float32) |
| 244 | + |
| 245 | + for b in range(num_train_batch): |
| 246 | + x = train_x[idx[b * batch_size:(b + 1) * batch_size]] |
| 247 | + x = augmentation(x, batch_size) |
| 248 | + y = train_y[idx[b * batch_size:(b + 1) * batch_size]] |
| 249 | + tx.copy_from_numpy(x) |
| 250 | + ty.copy_from_numpy(y) |
| 251 | + out = model.forward(tx) |
| 252 | + loss = autograd.softmax_cross_entropy(out, ty) |
| 253 | + train_correct += accuracy(tensor.to_numpy(out), y) |
| 254 | + train_loss += tensor.to_numpy(loss)[0] |
| 255 | + if DIST: |
| 256 | + if (spars == 0): |
| 257 | + sgd.backward_and_update(loss, threshold=50000) |
| 258 | + else: |
| 259 | + sgd.backward_and_sparse_update(loss, |
| 260 | + spars=spars, |
| 261 | + topK=topK, |
| 262 | + corr=corr) |
| 263 | + else: |
| 264 | + sgd(loss) |
| 265 | + |
| 266 | + if DIST: |
| 267 | + # Reduce the evaluation accuracy and loss from multiple devices |
| 268 | + reducer = tensor.Tensor((1,), dev, tensor.float32) |
| 269 | + train_correct = reduce_variable(train_correct, sgd, reducer) |
| 270 | + train_loss = reduce_variable(train_loss, sgd, reducer) |
| 271 | + |
| 272 | + # Output the training loss and accuracy |
| 273 | + if ((DIST == False) or (sgd.global_rank == 0)): |
| 274 | + print('Training loss = %f, training accuracy = %f' % |
| 275 | + (train_loss, train_correct / |
| 276 | + (num_train_batch * batch_size * world_size)), |
| 277 | + flush=True) |
| 278 | + |
| 279 | + # Evaluation phase |
| 280 | + autograd.training = False |
| 281 | + for b in range(num_test_batch): |
| 282 | + x = test_x[b * batch_size:(b + 1) * batch_size] |
| 283 | + y = test_y[b * batch_size:(b + 1) * batch_size] |
| 284 | + tx.copy_from_numpy(x) |
| 285 | + ty.copy_from_numpy(y) |
| 286 | + out_test = model.forward(tx) |
| 287 | + test_correct += accuracy(tensor.to_numpy(out_test), y) |
| 288 | + |
| 289 | + if DIST: |
| 290 | + # Reduce the evaulation accuracy from multiple devices |
| 291 | + test_correct = reduce_variable(test_correct, sgd, reducer) |
| 292 | + |
| 293 | + # Output the evaluation accuracy |
| 294 | + if ((DIST == False) or (sgd.global_rank == 0)): |
| 295 | + print('Evaluation accuracy = %f, Elapsed Time = %fs' % |
| 296 | + (test_correct / (num_test_batch * batch_size * world_size), |
| 297 | + time.time() - start_time), |
| 298 | + flush=True) |
| 299 | + |
| 300 | + |
| 301 | +if __name__ == '__main__': |
| 302 | + |
| 303 | + DIST = False |
| 304 | + train_mnist_cnn(DIST=DIST) |
0 commit comments