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# ============================================================================ 15 16import os 17import numpy as np 18 19from mindspore.communication.management import init 20from mindspore.communication.management import release 21from mindspore.communication.management import get_rank 22from mindspore.communication.management import get_group_size 23from mindspore.nn import Cell 24from mindspore.nn import ReLU 25from mindspore.nn import Dense 26from mindspore.nn import Flatten 27from mindspore.nn import Momentum 28import mindspore.ops.operations as P 29from mindspore.train.serialization import load_param_into_net 30from mindspore.train.callback import CheckpointConfig 31from mindspore.train.callback import ModelCheckpoint 32from mindspore.train.serialization import load_checkpoint 33 34from mindspore.nn import SoftmaxCrossEntropyWithLogits 35from mindspore.train import Model 36from mindspore.parallel import set_algo_parameters 37from mindspore import Tensor 38from mindspore.common.parameter import Parameter 39from mindspore import context 40from mindspore.context import ParallelMode 41 42context.set_context(mode=context.GRAPH_MODE, device_target='Ascend') 43 44def _count_unequal_element(data_expected, data_me, rtol, atol): 45 assert data_expected.shape == data_me.shape 46 total_count = len(data_expected.flatten()) 47 error = np.abs(data_expected - data_me) 48 greater = np.greater(error, atol + np.abs(data_me) * rtol) 49 loss_count = np.count_nonzero(greater) 50 assert (loss_count / total_count) < rtol, \ 51 "\ndata_expected_std:{0}\ndata_me_error:{1}\nloss:{2}". \ 52 format(data_expected[greater], data_me[greater], error[greater]) 53 54 55def allclose_nparray(data_expected, data_me, rtol, atol, equal_nan=True): 56 if np.any(np.isnan(data_expected)): 57 assert np.allclose(data_expected, data_me, rtol, atol, equal_nan=equal_nan) 58 elif not np.allclose(data_expected, data_me, rtol, atol, equal_nan=equal_nan): 59 _count_unequal_element(data_expected, data_me, rtol, atol) 60 else: 61 assert True 62 63def clean_all_ckpt_files(folder_path): 64 if os.path.exists(folder_path): 65 for file_name in os.listdir(folder_path): 66 if file_name.endswith('.ckpt') or file_name.endswith('.meta'): 67 os.remove(os.path.join(folder_path, file_name)) 68 69 70def find_newest_ckpt_file(folder_path): 71 ckpt_files = map(lambda f: os.path.join(folder_path, f), 72 filter(lambda f: f.endswith('.ckpt'), 73 os.listdir(folder_path))) 74 return max(ckpt_files, key=os.path.getctime) 75 76 77class FakeDataInitMode: 78 RandomInit = 0 79 OnesInit = 1 80 UniqueInit = 2 81 ZerosInit = 3 82 83 84class FakeData: 85 def __init__(self, size=1024, batch_size=32, image_size=(3, 224, 224), 86 num_classes=10, random_offset=0, use_parallel=False, 87 fakedata_mode=FakeDataInitMode.RandomInit): 88 self.size = size 89 self.rank_batch_size = batch_size 90 self.total_batch_size = self.rank_batch_size 91 self.random_offset = random_offset 92 self.image_size = image_size 93 self.num_classes = num_classes 94 self.rank_size = 1 95 self.rank_id = 0 96 self.batch_index = 0 97 self.image_data_type = np.float32 98 self.label_data_type = np.float32 99 self.is_onehot = True 100 self.fakedata_mode = fakedata_mode 101 102 if use_parallel is True: 103 init(backend_name='hccl') 104 self.rank_size = get_group_size() 105 self.rank_id = get_rank() 106 107 self.total_batch_size = self.rank_batch_size * self.rank_size 108 109 assert (self.size % self.total_batch_size) == 0 110 111 self.total_batch_data_size = (self.rank_size, self.rank_batch_size) + image_size 112 113 def get_dataset_size(self): 114 return int(self.size / self.total_batch_size) 115 116 def get_repeat_count(self): 117 return 1 118 119 def set_image_data_type(self, data_type): 120 self.image_data_type = data_type 121 122 def set_label_data_type(self, data_type): 123 self.label_data_type = data_type 124 125 def set_label_onehot(self, is_onehot=True): 126 self.is_onehot = is_onehot 127 128 def create_tuple_iterator(self, num_epochs=-1, do_copy=True): 129 _ = num_epochs 130 return self 131 132 def __getitem__(self, batch_index): 133 if batch_index * self.total_batch_size >= len(self): 134 raise IndexError("{} index out of range".format(self.__class__.__name__)) 135 rng_state = np.random.get_state() 136 np.random.seed(batch_index + self.random_offset) 137 if self.fakedata_mode == FakeDataInitMode.OnesInit: 138 img = np.ones(self.total_batch_data_size) 139 elif self.fakedata_mode == FakeDataInitMode.ZerosInit: 140 img = np.zeros(self.total_batch_data_size) 141 elif self.fakedata_mode == FakeDataInitMode.UniqueInit: 142 total_size = 1 143 for i in self.total_batch_data_size: 144 total_size = total_size * i 145 img = np.reshape(np.arange(total_size) * 0.0001, self.total_batch_data_size) 146 else: 147 img = np.random.randn(*self.total_batch_data_size) 148 target = np.random.randint(0, self.num_classes, size=(self.rank_size, self.rank_batch_size)) 149 np.random.set_state(rng_state) 150 img = img[self.rank_id] 151 target = target[self.rank_id] 152 img_ret = img.astype(self.image_data_type) 153 target_ret = target.astype(self.label_data_type) 154 if self.is_onehot: 155 target_onehot = np.zeros(shape=(self.rank_batch_size, self.num_classes)) 156 target_onehot[np.arange(self.rank_batch_size), target] = 1 157 target_ret = target_onehot.astype(self.label_data_type) 158 return Tensor(img_ret), Tensor(target_ret) 159 160 def __len__(self): 161 return self.size 162 163 def __iter__(self): 164 self.batch_index = 0 165 return self 166 167 def reset(self): 168 self.batch_index = 0 169 170 def __next__(self): 171 if self.batch_index * self.total_batch_size < len(self): 172 data = self[self.batch_index] 173 self.batch_index += 1 174 return data 175 raise StopIteration 176 177 178class OptimizerSemiAutoAndAutoParallel6Net(Cell): 179 def __init__(self, strategy_dict=None): 180 super().__init__() 181 shared_np = np.full((16, 1, 32, 32), 0.5, dtype=np.float32) 182 self.shared_weight = Parameter(Tensor(shared_np), name='shared_weight') 183 self.fc1 = Dense(in_channels=1024, 184 out_channels=116, 185 weight_init='ones', 186 bias_init='ones', 187 has_bias=True) 188 self.relu = ReLU() 189 self.sigmoid = P.Sigmoid() 190 self.add1 = P.Add() 191 self.add2 = P.Add() 192 self.mul1 = P.Mul().add_prim_attr('primitive_target', 'CPU') 193 self.mul2 = P.Mul() 194 self.mul3 = P.Mul() 195 self.flatten = Flatten() 196 197 mul2_weight_np = np.full((16, 116), 1, dtype=np.float32) 198 self.mul2_weight = Parameter(Tensor(mul2_weight_np), name='mul2_weight') 199 200 mul3_weight_np = np.full((16, 116), 1, dtype=np.float32) 201 self.mul3_weight = Parameter(Tensor(mul3_weight_np), name='mul3_weight') 202 203 if strategy_dict is not None: 204 self.add1.shard(strategy_dict['add1']) 205 self.mul1.shard(strategy_dict['mul1']) 206 self.fc1.matmul.shard(strategy_dict['fc1_matmul']) 207 self.fc1.bias_add.shard(strategy_dict['fc1_bias_add']) 208 self.mul2.shard(strategy_dict['mul2']) 209 self.mul3.shard(strategy_dict['mul3']) 210 211 def construct(self, inputs): 212 relu = self.relu(inputs) 213 sigmoid = self.sigmoid(inputs) 214 add1 = self.add1(relu, self.shared_weight) 215 mul = self.mul1(sigmoid, self.shared_weight) 216 add2 = self.add2(add1, mul) 217 flatten = self.flatten(add2) 218 dense = self.fc1(flatten) 219 mul2 = self.mul2(dense, self.mul2_weight) 220 out = self.mul3(mul2, self.mul3_weight) 221 return out 222 223 224class OptimizerSemiAutoAndAutoParallelFactory: 225 def __init__(self, net, strategy_dict=None): 226 self.parallel_ckpt = None 227 self.optimizer_parallel_ckpt = None 228 self.net = net 229 self.strategy_dict = strategy_dict 230 self.global_rank_id = None 231 self._set_parallel_env() 232 self._init_parallel() 233 234 def __enter__(self): 235 return self 236 237 def __exit__(self, exc_type, exc_val, exc_tb): 238 return 239 240 def __del__(self): 241 self._release_parallel() 242 243 def _set_parallel_env(self): 244 if 'RANK_ID' in os.environ: 245 self.global_rank_id = int(os.environ['RANK_ID']) 246 247 def _init_parallel(self): 248 self._init_parallel_flag = False 249 init(backend_name='hccl') 250 self._init_parallel_flag = True 251 252 def _release_parallel(self): 253 if self._init_parallel_flag: 254 release() 255 256 def _model_train_and_save_ckpt(self, net, dataset, epoch): 257 self.opt = Momentum(learning_rate=0.01, momentum=0.9, params=net.get_parameters()) 258 self.loss_fn = SoftmaxCrossEntropyWithLogits(reduction='mean') 259 self.model = Model(network=net, 260 loss_fn=self.loss_fn, 261 optimizer=self.opt) 262 ckpt_config = CheckpointConfig(keep_checkpoint_max=1) 263 ckpt_path = './rank_{}_ckpt'.format(self.global_rank_id) 264 ckpt_callback = ModelCheckpoint(prefix='parallel', directory=ckpt_path, 265 config=ckpt_config) 266 clean_all_ckpt_files(ckpt_path) 267 self.model.train(epoch=epoch, 268 train_dataset=dataset, 269 callbacks=[ckpt_callback], 270 dataset_sink_mode=False) 271 newest_ckpt_file = find_newest_ckpt_file(ckpt_path) 272 return load_checkpoint(newest_ckpt_file) 273 274 def mindspore_auto_parallel_impl(self, 275 dataset, 276 epoch, 277 device_num): 278 set_algo_parameters(fully_use_devices=False) 279 context.reset_auto_parallel_context() 280 context.set_auto_parallel_context(parallel_mode=ParallelMode.AUTO_PARALLEL, 281 device_num=device_num) 282 parallel_mode_net = self.net(self.strategy_dict) 283 self.parallel_ckpt = self._model_train_and_save_ckpt(net=parallel_mode_net, 284 dataset=dataset, epoch=epoch) 285 context.reset_auto_parallel_context() 286 287 def mindspore_optimizer_auto_parallel_impl(self, 288 dataset, 289 epoch, 290 device_num): 291 set_algo_parameters(fully_use_devices=False) 292 context.reset_auto_parallel_context() 293 context.set_auto_parallel_context(parallel_mode=ParallelMode.AUTO_PARALLEL, 294 device_num=device_num, 295 enable_parallel_optimizer=True) 296 parallel_mode_net = self.net(self.strategy_dict) 297 self.optimizer_parallel_ckpt = self._model_train_and_save_ckpt(net=parallel_mode_net, 298 dataset=dataset, epoch=epoch) 299 context.reset_auto_parallel_context() 300 301 def checkpoint_cmp(self, inputs_np): 302 optimizer_parallel_net = self.net(self.strategy_dict) 303 load_param_into_net(optimizer_parallel_net, self.optimizer_parallel_ckpt) 304 optimizer_parallel_out = optimizer_parallel_net(Tensor(inputs_np)) 305 306 parallel_net = self.net(self.strategy_dict) 307 load_param_into_net(parallel_net, self.parallel_ckpt) 308 parallel_out = parallel_net(Tensor(inputs_np)) 309 allclose_nparray(optimizer_parallel_out.asnumpy(), parallel_out.asnumpy(), 0.001, 0.001) 310 311def test_optimizer_parallel_auto_4p_6_parameter_same_strategy_1_1_2_1_momentum(): 312 inputs_np = np.random.randn(16, 1, 32, 32).astype(np.float32) 313 ds1 = FakeData(size=32, 314 batch_size=4, 315 image_size=(1, 32, 32), 316 use_parallel=True, 317 num_classes=116) 318 319 ds2 = FakeData(size=32, 320 batch_size=4, 321 image_size=(1, 32, 32), 322 use_parallel=True, 323 num_classes=116) 324 strategy_dict = {'add1': ((1, 1, 2, 1), (1, 1, 2, 1)), 325 'mul1': ((1, 1, 2, 1), (1, 1, 2, 1)), 326 'fc1_matmul': ((1, 2), (1, 2)), 327 'fc1_bias_add': ((1, 2), (2,)), 328 'mul2': ((1, 2), (1, 2)), 329 'mul3': ((1, 2), (1, 2))} 330 fact = OptimizerSemiAutoAndAutoParallelFactory(net=OptimizerSemiAutoAndAutoParallel6Net, 331 strategy_dict=strategy_dict) 332 fact.mindspore_auto_parallel_impl(dataset=ds1, epoch=2, device_num=4) 333 fact.mindspore_optimizer_auto_parallel_impl(dataset=ds2, epoch=2, device_num=4) 334 fact.checkpoint_cmp(inputs_np=inputs_np) 335