1# Copyright 2020 Huawei Technologies Co., Ltd 2# 3# Licensed under the Apache License, Version 2.0 (the "License"); 4# you may not use this file except in compliance with the License. 5# You may obtain a copy of the License at 6# 7# http://www.apache.org/licenses/LICENSE-2.0 8# 9# Unless required by applicable law or agreed to in writing, software 10# distributed under the License is distributed on an "AS IS" BASIS, 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12# See the License for the specific language governing permissions and 13# limitations under the License. 14# ============================================================================ 15import os 16import numpy as np 17 18from mindspore.communication.management import init 19from mindspore.communication.management import release 20from mindspore.communication.management import get_rank 21from mindspore.communication.management import get_group_size 22from mindspore.nn import Cell 23from mindspore.nn import Conv2d 24from mindspore.nn import ReLU 25from mindspore.nn import Dense 26from mindspore.nn import Softmax 27import mindspore.ops.operations as P 28from mindspore.train.serialization import load_param_into_net 29from mindspore.train.callback import CheckpointConfig 30from mindspore.train.callback import ModelCheckpoint 31from mindspore.train.serialization import load_checkpoint 32from mindspore.nn import Momentum 33from mindspore.nn import SoftmaxCrossEntropyWithLogits 34from mindspore.train import Model 35from mindspore.parallel import set_algo_parameters 36from mindspore.common.initializer import initializer 37from mindspore.common import dtype as mstype 38from mindspore import Tensor 39from mindspore.common.parameter import Parameter 40from mindspore import context 41from mindspore.context import ParallelMode 42 43context.set_context(mode=context.GRAPH_MODE, device_target='Ascend') 44 45 46def _count_unequal_element(data_expected, data_me, rtol, atol): 47 assert data_expected.shape == data_me.shape 48 total_count = len(data_expected.flatten()) 49 error = np.abs(data_expected - data_me) 50 greater = np.greater(error, atol + np.abs(data_me) * rtol) 51 loss_count = np.count_nonzero(greater) 52 assert (loss_count / total_count) < rtol, \ 53 "\ndata_expected_std:{0}\ndata_me_error:{1}\nloss:{2}". \ 54 format(data_expected[greater], data_me[greater], error[greater]) 55 56 57def allclose_nparray(data_expected, data_me, rtol, atol, equal_nan=True): 58 if np.any(np.isnan(data_expected)): 59 assert np.allclose(data_expected, data_me, rtol, atol, equal_nan=equal_nan) 60 elif not np.allclose(data_expected, data_me, rtol, atol, equal_nan=equal_nan): 61 _count_unequal_element(data_expected, data_me, rtol, atol) 62 else: 63 assert True 64 65 66def clean_all_ckpt_files(folder_path): 67 if os.path.exists(folder_path): 68 for file_name in os.listdir(folder_path): 69 if file_name.endswith('.ckpt') or file_name.endswith('.meta'): 70 os.remove(os.path.join(folder_path, file_name)) 71 72 73def find_newest_ckpt_file(folder_path): 74 ckpt_files = map(lambda f: os.path.join(folder_path, f), 75 filter(lambda f: f.endswith('.ckpt'), 76 os.listdir(folder_path))) 77 return max(ckpt_files, key=os.path.getctime) 78 79 80class FakeDataInitMode: 81 RandomInit = 0 82 OnesInit = 1 83 UniqueInit = 2 84 ZerosInit = 3 85 86 87class FakeData: 88 def __init__(self, size=1024, batch_size=32, image_size=(3, 224, 224), 89 num_classes=10, random_offset=0, use_parallel=False, 90 fakedata_mode=FakeDataInitMode.RandomInit): 91 self.size = size 92 self.rank_batch_size = batch_size 93 self.total_batch_size = self.rank_batch_size 94 self.random_offset = random_offset 95 self.image_size = image_size 96 self.num_classes = num_classes 97 self.rank_size = 1 98 self.rank_id = 0 99 self.batch_index = 0 100 self.image_data_type = np.float32 101 self.label_data_type = np.float32 102 self.is_onehot = True 103 self.fakedata_mode = fakedata_mode 104 105 if use_parallel is True: 106 init(backend_name='hccl') 107 self.rank_size = get_group_size() 108 self.rank_id = get_rank() 109 110 self.total_batch_size = self.rank_batch_size * self.rank_size 111 112 assert (self.size % self.total_batch_size) == 0 113 114 self.total_batch_data_size = (self.rank_size, self.rank_batch_size) + image_size 115 116 def get_dataset_size(self): 117 return int(self.size / self.total_batch_size) 118 119 def get_repeat_count(self): 120 return 1 121 122 def set_image_data_type(self, data_type): 123 self.image_data_type = data_type 124 125 def set_label_data_type(self, data_type): 126 self.label_data_type = data_type 127 128 def set_label_onehot(self, is_onehot=True): 129 self.is_onehot = is_onehot 130 131 def create_tuple_iterator(self, num_epochs=-1, do_copy=True): 132 _ = num_epochs 133 return self 134 135 def __getitem__(self, batch_index): 136 if batch_index * self.total_batch_size >= len(self): 137 raise IndexError("{} index out of range".format(self.__class__.__name__)) 138 rng_state = np.random.get_state() 139 np.random.seed(batch_index + self.random_offset) 140 if self.fakedata_mode == FakeDataInitMode.OnesInit: 141 img = np.ones(self.total_batch_data_size) 142 elif self.fakedata_mode == FakeDataInitMode.ZerosInit: 143 img = np.zeros(self.total_batch_data_size) 144 elif self.fakedata_mode == FakeDataInitMode.UniqueInit: 145 total_size = 1 146 for i in self.total_batch_data_size: 147 total_size = total_size * i 148 img = np.reshape(np.arange(total_size) * 0.0001, self.total_batch_data_size) 149 else: 150 img = np.random.randn(*self.total_batch_data_size) 151 target = np.random.randint(0, self.num_classes, size=(self.rank_size, self.rank_batch_size)) 152 np.random.set_state(rng_state) 153 img = img[self.rank_id] 154 target = target[self.rank_id] 155 img_ret = img.astype(self.image_data_type) 156 target_ret = target.astype(self.label_data_type) 157 if self.is_onehot: 158 target_onehot = np.zeros(shape=(self.rank_batch_size, self.num_classes)) 159 target_onehot[np.arange(self.rank_batch_size), target] = 1 160 target_ret = target_onehot.astype(self.label_data_type) 161 return Tensor(img_ret), Tensor(target_ret) 162 163 def __len__(self): 164 return self.size 165 166 def __iter__(self): 167 self.batch_index = 0 168 return self 169 170 def reset(self): 171 self.batch_index = 0 172 173 def __next__(self): 174 if self.batch_index * self.total_batch_size < len(self): 175 data = self[self.batch_index] 176 self.batch_index += 1 177 return data 178 raise StopIteration 179 180 181class ParallelStrategySearchNet(Cell): 182 def __init__(self, in_channel, out_channel, axis, input_shape, mul_size, 183 test_size, prelu_size, transpose_b, matmul_size, num_class): 184 super().__init__() 185 mul_np = np.full(mul_size, 0.5, dtype=np.float32) 186 self.mul_weight = Parameter(Tensor(mul_np), name="mul_weight") 187 bias_np = np.full((12,), 7.1, dtype=np.float32) 188 self.bias = Parameter(Tensor(bias_np), name="bias") 189 prelu_np = np.full(prelu_size, 0.8, dtype=np.float32) 190 self.prelu_weight = Parameter(Tensor(prelu_np), name="prelu_weight") 191 matmul_np = np.full(matmul_size, 1.1, dtype=np.float32) 192 self.matmul_weight = Parameter(Tensor(matmul_np), name="matmul_weight") 193 self.mul = P.Mul() 194 self.conv = Conv2d(in_channels=in_channel, out_channels=out_channel, 195 kernel_size=5, has_bias=True, 196 weight_init='ones', bias_init='ones', 197 pad_mode='valid') 198 self.conv.conv2d.shard(((8, 1, 1, 1), (1, 1, 1, 1))) 199 self.scalar = 0.5 200 self.parameter = Parameter( 201 initializer(0.5, test_size, dtype=mstype.float32), 202 name='parameter') 203 self.tensor = Tensor(np.full(test_size, 0.05, dtype=np.float32)) 204 self.softmax = Softmax(axis=axis) 205 self.relu = ReLU() 206 self.relu.relu.add_prim_attr("primitive_target", "CPU") 207 self.reshape = P.Reshape() 208 self.input_shape = input_shape 209 self.equal = P.Equal() 210 self.cast = P.Cast() 211 self.concat = P.Concat(axis=1) 212 self.reduce_sum = P.ReduceSum() 213 self.bias_add = P.BiasAdd() 214 self.cos = P.Cos() 215 self.prelu = P.PReLU() 216 self.matmul = P.MatMul(transpose_b=transpose_b) 217 self.l2norm = P.L2Normalize(axis=(1 - axis)) 218 self.tensoradd = P.Add() 219 self.strided_slice = P.StridedSlice() 220 self.dense = Dense(in_channels=6, 221 out_channels=num_class, 222 weight_init='ones', 223 bias_init='ones', 224 has_bias=True) 225 226 def construct(self, inputs): 227 x = self.conv(inputs) 228 x = self.softmax(x) 229 x = self.relu(x) 230 x = self.mul(x, self.mul_weight) 231 x = self.reshape(x, self.input_shape) 232 y = self.parameter * self.tensor * self.scalar 233 z = self.equal(self.parameter, self.scalar) 234 z = self.cast(z, mstype.float16) 235 z = self.cast(z, mstype.float32) 236 x = self.concat((x, y, z)) 237 x = self.reduce_sum(x, (2, 3)) 238 x = self.bias_add(x, self.bias) 239 y = self.cos(x) 240 y = self.prelu(y, self.prelu_weight) 241 z = self.matmul(x, self.matmul_weight) 242 z = self.l2norm(z) 243 x = self.tensoradd(y, z) 244 x = self.strided_slice(x, (0, 0), (32, 6), (1, 1)) 245 x = self.dense(x) 246 return x 247 248 249class ParallelStrategySearchFactory: 250 def __init__(self, standalone_mode_net, parallel_mode_net): 251 self.standalone_mode_net = standalone_mode_net 252 self.parallel_mode_net = parallel_mode_net 253 self.parallel_ckpt = None 254 self.standalone_ckpt = None 255 self.global_rank_id = None 256 self._set_parallel_env() 257 self._init_parallel() 258 259 def __enter__(self): 260 return self 261 262 def __exit__(self, exc_type, exc_val, exc_tb): 263 return 264 265 def __del__(self): 266 self._release_parallel() 267 268 def _set_parallel_env(self): 269 if 'RANK_ID' in os.environ: 270 self.global_rank_id = int(os.environ['RANK_ID']) 271 272 def _init_parallel(self): 273 self._init_parallel_flag = False 274 init(backend_name='hccl') 275 self._init_parallel_flag = True 276 277 def _release_parallel(self): 278 if self._init_parallel_flag: 279 release() 280 281 def _model_train_and_save_ckpt(self, net, dataset, epoch): 282 self.opt = Momentum(learning_rate=0.01, momentum=0.9, params=net.get_parameters()) 283 self.loss_fn = SoftmaxCrossEntropyWithLogits(reduction='mean') 284 self.model = Model(network=net, 285 loss_fn=self.loss_fn, 286 optimizer=self.opt) 287 ckpt_config = CheckpointConfig(keep_checkpoint_max=1) 288 ckpt_path = './rank_{}_ckpt'.format(self.global_rank_id) 289 ckpt_callback = ModelCheckpoint(prefix='parallel', directory=ckpt_path, 290 config=ckpt_config) 291 clean_all_ckpt_files(ckpt_path) 292 self.model.train(epoch=epoch, 293 train_dataset=dataset, 294 callbacks=[ckpt_callback], 295 dataset_sink_mode=False) 296 newest_ckpt_file = find_newest_ckpt_file(ckpt_path) 297 return load_checkpoint(newest_ckpt_file) 298 299 def mindspore_auto_parallel_impl(self, dataset, epoch, device_num, auto_parallel_search_mode="dynamic_programming"): 300 parallel_mode_net = self.parallel_mode_net 301 set_algo_parameters(fully_use_devices=False) 302 context.reset_auto_parallel_context() 303 context.set_auto_parallel_context(parallel_mode=ParallelMode.AUTO_PARALLEL, 304 device_num=device_num, 305 auto_parallel_search_mode=auto_parallel_search_mode) 306 self.parallel_ckpt = self._model_train_and_save_ckpt(net=parallel_mode_net, 307 dataset=dataset, epoch=epoch) 308 context.reset_auto_parallel_context() 309 310 def mindspore_standalone_impl(self, dataset, epoch): 311 standalone_mode_net = self.standalone_mode_net 312 context.reset_auto_parallel_context() 313 context.set_auto_parallel_context(parallel_mode=ParallelMode.STAND_ALONE) 314 self.standalone_ckpt = self._model_train_and_save_ckpt(net=standalone_mode_net, 315 dataset=dataset, epoch=epoch) 316 context.reset_auto_parallel_context() 317 318 def checkpoint_cmp(self, inputs_np): 319 standalone_net = self.standalone_mode_net 320 load_param_into_net(standalone_net, self.standalone_ckpt) 321 standalone_out = standalone_net(Tensor(inputs_np)) 322 323 parallel_net = self.standalone_mode_net 324 load_param_into_net(parallel_net, self.parallel_ckpt) 325 parallel_out = parallel_net(Tensor(inputs_np)) 326 allclose_nparray(standalone_out.asnumpy(), parallel_out.asnumpy(), 327 0.001, 0.001) 328 329 330def test_auto_parallel_strategy_search_axis_1_basic(): 331 inputs_np = np.random.randn(32, 3, 224, 224).astype(np.float32) 332 standalone_mode_net = ParallelStrategySearchNet(in_channel=3, 333 out_channel=8, axis=1, input_shape=(32, 4, 110, -1), 334 mul_size=(32, 1, 220, 220), test_size=(32, 4, 110, 880), 335 prelu_size=(1,), transpose_b=True, matmul_size=(1, 12), 336 num_class=12) 337 context.reset_auto_parallel_context() 338 context.set_auto_parallel_context(parallel_mode=ParallelMode.AUTO_PARALLEL) 339 parallel_mode_net = ParallelStrategySearchNet(in_channel=3, 340 out_channel=8, axis=1, input_shape=(32, 4, 110, -1), 341 mul_size=(32, 1, 220, 220), test_size=(32, 4, 110, 880), 342 prelu_size=(1,), transpose_b=True, matmul_size=(1, 12), 343 num_class=12) 344 parallel_mode_net.cos.shard(((2, 4),)) 345 parallel_mode_net.matmul.shard(((1, 2), (1, 2))) 346 standalone_dataset = FakeData(size=128, batch_size=32, 347 image_size=(3, 224, 224), num_classes=12) 348 fact = ParallelStrategySearchFactory(standalone_mode_net=standalone_mode_net, 349 parallel_mode_net=parallel_mode_net) 350 fact.mindspore_standalone_impl(dataset=standalone_dataset, epoch=2) 351 parallel_dataset = FakeData(size=128, batch_size=4, 352 image_size=(3, 224, 224), use_parallel=True, 353 num_classes=12) 354 fact.mindspore_auto_parallel_impl(dataset=parallel_dataset, 355 epoch=2, device_num=8) 356 fact.checkpoint_cmp(inputs_np=inputs_np) 357 358 359def test_auto_parallel_recursive_strategy_search_axis_1_basic(): 360 inputs_np = np.random.randn(32, 3, 224, 224).astype(np.float32) 361 standalone_mode_net = ParallelStrategySearchNet(in_channel=3, 362 out_channel=8, axis=1, input_shape=(32, 4, 110, -1), 363 mul_size=(32, 1, 220, 220), test_size=(32, 4, 110, 880), 364 prelu_size=(1,), transpose_b=True, matmul_size=(1, 12), 365 num_class=12) 366 context.reset_auto_parallel_context() 367 context.set_auto_parallel_context(parallel_mode=ParallelMode.AUTO_PARALLEL) 368 parallel_mode_net = ParallelStrategySearchNet(in_channel=3, 369 out_channel=8, axis=1, input_shape=(32, 4, 110, -1), 370 mul_size=(32, 1, 220, 220), test_size=(32, 4, 110, 880), 371 prelu_size=(1,), transpose_b=True, matmul_size=(1, 12), 372 num_class=12) 373 standalone_dataset = FakeData(size=128, batch_size=32, 374 image_size=(3, 224, 224), num_classes=12) 375 fact = ParallelStrategySearchFactory(standalone_mode_net=standalone_mode_net, 376 parallel_mode_net=parallel_mode_net) 377 fact.mindspore_standalone_impl(dataset=standalone_dataset, epoch=2) 378 parallel_dataset = FakeData(size=128, batch_size=4, 379 image_size=(3, 224, 224), use_parallel=True, 380 num_classes=12) 381 fact.mindspore_auto_parallel_impl(dataset=parallel_dataset, 382 epoch=2, device_num=8, auto_parallel_search_mode="recursive_programming") 383 fact.checkpoint_cmp(inputs_np=inputs_np) 384