From b60e54e0d54caf3913cf0ebfc7520ce7c11a4a4b Mon Sep 17 00:00:00 2001 From: yao_yf Date: Fri, 4 Mar 2022 15:05:28 +0800 Subject: [PATCH] support not only power of 2 --- .../ccsrc/frontend/parallel/device_manager.cc | 89 ++++++++- .../ccsrc/frontend/parallel/device_manager.h | 5 +- .../parallel/ops_info/activation_info.cc | 8 +- .../parallel/ops_info/batchnorm_info.cc | 8 +- .../frontend/parallel/ops_info/conv2d_info.cc | 4 +- .../parallel/ops_info/crop_and_resize_info.cc | 5 +- .../frontend/parallel/ops_info/gather_info.cc | 9 +- .../parallel/ops_info/gatherd_info.cc | 2 +- .../frontend/parallel/ops_info/matmul_info.cc | 152 ++++++++++----- .../frontend/parallel/ops_info/matmul_info.h | 4 + .../parallel/ops_info/operator_info.cc | 157 ++++++++++++--- .../parallel/ops_info/operator_info.h | 16 +- .../parallel/ops_info/reduce_method_info.cc | 8 +- .../parallel/ops_info/reshape_info.cc | 2 +- .../parallel/ops_info/roi_align_info.cc | 4 +- .../frontend/parallel/ops_info/slice_info.cc | 2 +- .../parallel/ops_info/strided_slice_info.cc | 2 +- .../parallel/ops_info/tensordot_info.cc | 2 +- .../frontend/parallel/ops_info/tile_info.cc | 2 +- .../frontend/parallel/ops_info/topk_info.cc | 2 +- .../ops_info/unsorted_segment_op_info.cc | 4 +- .../frontend/parallel/parameter_manager.cc | 15 +- .../pipeline_transformer.cc | 12 +- .../frontend/parallel/step_auto_parallel.cc | 6 + .../ccsrc/frontend/parallel/step_parallel.cc | 14 +- .../parallel/tensor_layout/arrangement.cc | 4 +- .../tensor_layout/construct_operator.cc | 13 +- .../tensor_layout/construct_operator.h | 3 +- .../redistribution_layout_transfer.cc | 4 +- .../redistribution_operator_infer.cc | 2 +- .../parallel/tensor_layout/shape_util.cc | 2 +- mindspore/python/mindspore/nn/optim/adasum.py | 2 - .../tensor_layout/construct_operator_test.cc | 2 +- .../test_device_num_not_power_of_2.py | 181 ++++++++++++++++++ .../test_mix_precision_hybrid_parallel.py | 2 +- .../parallel/test_pipeline_opt_detection.py | 72 +++---- .../test_pipeline_shared_parameter.py | 12 +- .../ut/python/parallel/test_pipeline_split.py | 142 +++++++++----- tests/ut/python/parallel/test_tile.py | 6 +- 39 files changed, 756 insertions(+), 225 deletions(-) create mode 100644 tests/ut/python/parallel/test_device_num_not_power_of_2.py diff --git a/mindspore/ccsrc/frontend/parallel/device_manager.cc b/mindspore/ccsrc/frontend/parallel/device_manager.cc index 456e25c5636..a84b3bf0f8a 100644 --- a/mindspore/ccsrc/frontend/parallel/device_manager.cc +++ b/mindspore/ccsrc/frontend/parallel/device_manager.cc @@ -19,6 +19,7 @@ #include #include #include +#include #include "utils/hash_set.h" #include "frontend/parallel/step_parallel.h" @@ -160,6 +161,67 @@ std::shared_ptr GetListMemberByIndex(size_t index, const std::vector= 4 && rank_list[0] >= 4) { + return SUCCESS; + } + } + return FAILED; +} + +Status IsFeasibleDeiveList(const RankList &rank_list) { + std::unordered_map server_ranks_map; + for (auto rank : rank_list) { + int64_t server_id = rank / NODE_PER_SERVER; + int64_t local_rank = rank % NODE_PER_SERVER; + server_ranks_map[server_id].push_back(local_rank); + } + std::vector server_ranks_list; + std::transform(server_ranks_map.begin(), server_ranks_map.end(), std::back_inserter(server_ranks_list), + [](auto pairs) { return pairs.second; }); + auto server0_local_ranks = server_ranks_list[0]; + bool is_all_server_same_count = + std::all_of(server_ranks_list.begin(), server_ranks_list.end(), + [&server0_local_ranks](auto ranks) { return ranks == server0_local_ranks; }); + if (!is_all_server_same_count) { + MS_LOG(INFO) << "All server should has the same ranks, which means rank_id % 8 in each server should be the same. " + "current rank list is" + << rank_list; + return FAILED; + } + return IsFeasibleDeiveListOneServer(server0_local_ranks); +} +} // namespace + +Status DeviceManager::CheckDeviceList(const RankList &rank_list) { + auto ms_context = MsContext::GetInstance(); + MS_EXCEPTION_IF_NULL(ms_context); + std::string backend = ms_context->get_param(MS_CTX_DEVICE_TARGET); + if (backend == kAscendDevice || backend == kDavinciDevice) { + return IsFeasibleDeiveList(rank_list); + } + return SUCCESS; +} + // E.g. devices = [0, 1, 2, 3, 4, 5, 6, 7], stage_map = [4, 4], // therefore the stage_devices_ = [[0, 1, 2, 3], [4, 5, 6, 7]]. Status DeviceManager::Init(const RankList &devices, int64_t global_device_rank, const RankList &stage_map, @@ -357,23 +419,32 @@ std::string DeviceManager::GenerateGroupNameByRanks(RankList ranks) { // Create the group with the given devices and the given name. The GroupManager // gm_ will create a new group only if there does not exit a group with the same // name. Otherwise, let the pointer g point to that group. -Group DeviceManager::CreateGroup(const std::string &group_name, - const std::vector &devices) { - Group g; - (void)gm_.CreateGroup(group_name, devices, &g); - return g; +Status DeviceManager::CreateGroup(const std::string &group_name, + const std::vector &devices, Group *const comm_group) { + RankList rank_list; + std::transform(devices.begin(), devices.end(), std::back_inserter(rank_list), + [](Device device) { return device.rank(); }); + if (CheckDeviceList(rank_list) != SUCCESS) { + MS_LOG(ERROR) << "Create communication group failed, the rank list is: " << rank_list; + return FAILED; + } + return gm_.CreateGroup(group_name, devices, comm_group); } // Create the group with only the given devices' ranks. -Group DeviceManager::CreateGroup(const RankList &dev_ranks) { +Status DeviceManager::CreateGroup(const RankList &dev_ranks, Group *const comm_group) { mindspore::HashSet rank_set(dev_ranks.begin(), dev_ranks.end()); if (dev_ranks.size() != rank_set.size()) { - MS_LOG(EXCEPTION) << "Invalid dev ranks(" << dev_ranks << "), it has the Duplicate elements in list"; + MS_LOG(ERROR) << "Invalid dev ranks(" << dev_ranks << "), it has the Duplicate elements in list"; + return FAILED; + } + if (CheckDeviceList(dev_ranks) != SUCCESS) { + MS_LOG(ERROR) << "Create communication group failed, the rank list is: " << dev_ranks; + return FAILED; } - std::string group_name = GenerateGroupNameByRanks(dev_ranks); auto dev_list = CreateDeviceListByRankList(dev_ranks); - return CreateGroup(group_name, dev_list); + return CreateGroup(group_name, dev_list, comm_group); } void DeviceManager::Clear() { diff --git a/mindspore/ccsrc/frontend/parallel/device_manager.h b/mindspore/ccsrc/frontend/parallel/device_manager.h index 7c40cbdfdc2..01ed6e7798f 100644 --- a/mindspore/ccsrc/frontend/parallel/device_manager.h +++ b/mindspore/ccsrc/frontend/parallel/device_manager.h @@ -71,8 +71,8 @@ class DeviceManager { Device CreateNewDeviceByRank(int64_t rank) const; std::vector CreateDeviceListByRankList(RankList ranks); std::string GenerateGroupNameByRanks(RankList dev_ranks); - Group CreateGroup(const std::string &group_name, const std::vector &devices); - Group CreateGroup(const RankList &dev_ranks); + Status CreateGroup(const std::string &group_name, const std::vector &devices, Group *const comm_group); + Status CreateGroup(const RankList &dev_ranks, Group *const comm_group); size_t DeviceNum() const { return devices_.size(); } int64_t stage_num() const { return stage_num_; } @@ -91,6 +91,7 @@ class DeviceManager { std::vector>> group_info() const { return gm_.group_info(); } std::string FindRankListNameByHashName(const std::string &hash_name); RankList FindRankListByHashName(const std::string &hash_name); + Status CheckDeviceList(const RankList &rank_list); private: std::vector> devices_; diff --git a/mindspore/ccsrc/frontend/parallel/ops_info/activation_info.cc b/mindspore/ccsrc/frontend/parallel/ops_info/activation_info.cc index 446f3c85b1f..42a39b3b5cb 100644 --- a/mindspore/ccsrc/frontend/parallel/ops_info/activation_info.cc +++ b/mindspore/ccsrc/frontend/parallel/ops_info/activation_info.cc @@ -280,7 +280,7 @@ Status CumSumInfo::InferMirrorOps() { Shape input_a_tensor_map = inputs_tensor_map_.at(0); std::vector input_a_group; if (CreateGroupByTensorMap(input_a_tensor_map, &input_a_group) != SUCCESS) { - MS_LOG(ERROR) << name_ << ": Create group for input a failed."; + ReportError(name_ + ": Create group for input a failed."); return FAILED; } OperatorVector op_for_input_a, op_for_axis; @@ -315,7 +315,7 @@ Status ActivationBase::InferMirrorOps() { Shape tensor_map = inputs_tensor_map_[0]; std::vector group; if (CreateGroupByTensorMap(tensor_map, &group) != SUCCESS) { - MS_LOG(ERROR) << name_ << " : Create group failed."; + ReportError(name_ + ": Create group failed."); return FAILED; } @@ -423,7 +423,7 @@ Status CastInfo::InferMirrorOps() { Shape tensor_map = inputs_tensor_map_[0]; std::vector group; if (CreateGroupByTensorMap(tensor_map, &group) != SUCCESS) { - MS_LOG(ERROR) << name_ << " : Create group failed."; + ReportError(name_ + ": Create group failed."); return FAILED; } @@ -515,7 +515,7 @@ Status ExpandDimsInfo::InferMirrorOps() { std::vector group; if (CreateGroupByTensorMap(inputs_tensor_map_[0], &group) != SUCCESS) { - MS_LOG(ERROR) << name_ << ": Create group failed"; + ReportError(name_ + ": Create group failed."); return FAILED; } diff --git a/mindspore/ccsrc/frontend/parallel/ops_info/batchnorm_info.cc b/mindspore/ccsrc/frontend/parallel/ops_info/batchnorm_info.cc index 6f4ece2acb8..8054821a01d 100644 --- a/mindspore/ccsrc/frontend/parallel/ops_info/batchnorm_info.cc +++ b/mindspore/ccsrc/frontend/parallel/ops_info/batchnorm_info.cc @@ -170,7 +170,11 @@ Status BatchNormInfo::InferAllReduceGroupBySize() { } MS_LOG(INFO) << name_ << ": The group rank list is " << group_rank_list; - Group g = g_device_manager->CreateGroup(group_rank_list); + Group g; + if (g_device_manager->CreateGroup(group_rank_list, &g) != SUCCESS) { + MS_LOG(ERROR) << "The node " << cnode_->fullname_with_scope() << " create sync allreduce failed"; + return FAILED; + } forward_allreduce_group_.push_back(g); return SUCCESS; } @@ -224,7 +228,7 @@ Status BatchNormInfo::InferForwardCommunication() { std::vector group_list; if (CreateGroupByTensorMap(tmp_map, &group_list) != SUCCESS) { - MS_LOG(ERROR) << name_ << ": Create group failed"; + ReportError(name_ + ": Create group failed."); return FAILED; } diff --git a/mindspore/ccsrc/frontend/parallel/ops_info/conv2d_info.cc b/mindspore/ccsrc/frontend/parallel/ops_info/conv2d_info.cc index 00d877ff2ae..94db178195e 100644 --- a/mindspore/ccsrc/frontend/parallel/ops_info/conv2d_info.cc +++ b/mindspore/ccsrc/frontend/parallel/ops_info/conv2d_info.cc @@ -651,7 +651,7 @@ Status Conv2DInfo::InferForwardCommunication() { std::vector group_list; if (CreateGroupByDim(relevant_dim_index, &group_list) != SUCCESS) { - MS_LOG(ERROR) << name_ << ": Create group failed"; + ReportError(name_ + ": Create group failed"); return FAILED; } @@ -1080,7 +1080,7 @@ Status Conv2DBackpropInputInfo::InferMirrorOps() { for (size_t i = 0; i < inputs_tensor_map_.size(); ++i) { std::vector group; if (CreateGroupByTensorMap(inputs_tensor_map_[i], &group) != SUCCESS) { - MS_LOG(ERROR) << name_ << ": Create group failed, the input index is " << i; + ReportError(name_ + ": Create group failed, the input index is " + std::to_string(i)); mirror_ops_.clear(); return FAILED; } diff --git a/mindspore/ccsrc/frontend/parallel/ops_info/crop_and_resize_info.cc b/mindspore/ccsrc/frontend/parallel/ops_info/crop_and_resize_info.cc index cc5a22d15f3..02b51334e35 100644 --- a/mindspore/ccsrc/frontend/parallel/ops_info/crop_and_resize_info.cc +++ b/mindspore/ccsrc/frontend/parallel/ops_info/crop_and_resize_info.cc @@ -121,7 +121,10 @@ Status CropAndResizeInfo::InferGroup() { } MS_LOG(INFO) << name_ << ": The group rank is " << group_devices; - group_ = g_device_manager->CreateGroup(group_devices); + if (g_device_manager->CreateGroup(group_devices, &group_) != SUCCESS) { + MS_LOG(ERROR) << "The node " << cnode_->fullname_with_scope() << " create sync allreduce failed"; + return FAILED; + } return SUCCESS; } diff --git a/mindspore/ccsrc/frontend/parallel/ops_info/gather_info.cc b/mindspore/ccsrc/frontend/parallel/ops_info/gather_info.cc index 29d10781447..05927941d60 100644 --- a/mindspore/ccsrc/frontend/parallel/ops_info/gather_info.cc +++ b/mindspore/ccsrc/frontend/parallel/ops_info/gather_info.cc @@ -322,7 +322,7 @@ Status GatherInfo::CheckStrategy(const StrategyPtr &strategy) { auto param_strategy = strategy->GetInputDim().at(0); auto slice_shape = param_shape.at(param_shape.size() - 1) / param_strategy.at(param_strategy.size() - 1); if ((target_ != CPU) && (slice_shape % 8 != 0) && (slice_shape != 1)) { - MS_LOG(ERROR) << name_ << ": Last dim of param slice shape need 32Byte aligned."; + ReportError(name_ + ": Last dim of param slice shape need 32Byte aligned."); return FAILED; } @@ -445,7 +445,7 @@ Status GatherInfo::InferMirrorOps() { Shape input_a_tensor_map = inputs_tensor_map_.at(0); std::vector input_a_group; if (CreateGroupByTensorMap(input_a_tensor_map, &input_a_group) != SUCCESS) { - MS_LOG(ERROR) << name_ << " : Create group for input a failed."; + ReportError(name_ + " : Create group for input a failed."); return FAILED; } @@ -782,7 +782,10 @@ Status GatherInfo::InferGroup() { } MS_LOG(INFO) << name_ << ": The group ranks is " << group_devices; - group_ = g_device_manager->CreateGroup(group_devices); + if (g_device_manager->CreateGroup(group_devices, &group_) != SUCCESS) { + MS_LOG(ERROR) << "The node " << cnode_->fullname_with_scope() << " create reduce group failed in table row split."; + return FAILED; + } return SUCCESS; } diff --git a/mindspore/ccsrc/frontend/parallel/ops_info/gatherd_info.cc b/mindspore/ccsrc/frontend/parallel/ops_info/gatherd_info.cc index dbe27c698bf..85374448e8e 100644 --- a/mindspore/ccsrc/frontend/parallel/ops_info/gatherd_info.cc +++ b/mindspore/ccsrc/frontend/parallel/ops_info/gatherd_info.cc @@ -154,7 +154,7 @@ Status GatherDInfo::InferMirrorOps() { std::vector group; if (CreateGroupByTensorMap(inputs_tensor_map_[0], &group) != SUCCESS) { - MS_LOG(ERROR) << name_ << ": Create group failed"; + ReportError(name_ + ": Create group failed."); return FAILED; } diff --git a/mindspore/ccsrc/frontend/parallel/ops_info/matmul_info.cc b/mindspore/ccsrc/frontend/parallel/ops_info/matmul_info.cc index e5dcbd026d6..7ec574bcaa7 100644 --- a/mindspore/ccsrc/frontend/parallel/ops_info/matmul_info.cc +++ b/mindspore/ccsrc/frontend/parallel/ops_info/matmul_info.cc @@ -262,7 +262,7 @@ Status MatMulBase::InferForwardCommunication() { std::vector group_list; if (CreateGroupByDim(relevant_dimension_index, &group_list) != SUCCESS) { - MS_LOG(ERROR) << name_ << " : Infer forward communication, create group failed."; + ReportError(name_ + " : Infer forward communication, create group failed."); return FAILED; } else if (group_list.empty()) { MS_LOG(INFO) << name_ << " : Forward all reduce is not required."; @@ -405,6 +405,92 @@ Status MatMulBase::SwapLastTwoElements(mindspore::parallel::Shape *const input) return SUCCESS; } +Status MatMulBase::GenerateStrategiesBase(int64_t stage_id, size_t dev_num, const Shape &input0_shape, + Shape input1_shape, std::vector *const sp_vector) { + // The shape of input0 (input1) + // E.g., input0 = [100, 200, 300], input1 = [300, 400] + + // Combining the input0_shape and input1_shape + // E.g., combined_shape = [100, 200, 300, 400] + size_t input1_shape_size = input1_shape.size(), input0_shape_size = input0_shape.size(); + Dimensions combined_partitions; + Shape combined_shape; + // In SwapLastTwoElements(), it is guaranteed that input0_shape.size() and input1_shape.size() are both larger than 2 + if (input0_shape.size() >= input1_shape.size()) { + combined_shape = input0_shape; + combined_shape.push_back(input1_shape[input1_shape.size() - 1]); + } else { + combined_shape = input1_shape; + combined_shape.push_back(input0_shape[input0_shape.size() - 2]); + } + std::function recursive = [&stage_id, &dev_num, &sp_vector, &combined_partitions, + &combined_shape, &input1_shape_size, &recursive, + &input0_shape_size, this](uint64_t current_index, size_t n) { + // Finishing the recursive steps, if the strategy is valid, then calculate the cost + // for this operator under the strategy. + if (current_index == combined_shape.size()) { + StrategyPtr sp; + if (this->PrepareStrategy(stage_id, dev_num, combined_partitions, input0_shape_size, input1_shape_size, &sp) == + FAILED) { + return; + } + sp_vector->push_back(sp); + } else { + MS_LOG(DEBUG) << name_ << " : The value input0_shape_size: " << input0_shape_size + << ", input1_shape_size: " << input1_shape_size; + for (uint64_t i = 1; i <= n; i *= 2) { + if (n % i == 0 && LongToSize(combined_shape[current_index]) % i == 0) { + combined_partitions.push_back(i); + recursive(current_index + 1, n / i); + combined_partitions.pop_back(); + } + } + } + }; + recursive(0, dev_num); + if (sp_vector->empty()) { + MS_LOG(EXCEPTION) << name_ << " : No available strategy."; + } + return Status::SUCCESS; +} + +Status MatMulBase::GenerateStrategiesNotPower2(int64_t stage_id, size_t dev_num_not_2_power, + const std::vector &sp_vector_2_power_part) { + std::vector sp_vector; + size_t related_dim_left = transpose_a_ ? inputs_shape_[0].size() - 2 : inputs_shape_[0].size() - 1; + size_t related_dim_right = transpose_b_ ? inputs_shape_[1].size() - 1 : inputs_shape_[1].size() - 2; + // Handle the not power of 2 part. + for (auto &stra : sp_vector_2_power_part) { + auto stra_arrays = stra->GetInputDim(); + if (stra_arrays.size() != 2) { + MS_LOG(ERROR) << "The generated strategy of matmul dose not match two input, the strategy is: " << stra_arrays; + } + for (size_t i = 0; i < 2; ++i) { + size_t stra_size = stra_arrays[i].size(); + for (size_t j = 0; j < stra_size; ++j) { + if (i == 1 && j == related_dim_right) { + continue; + } + auto new_stra_arrays{stra_arrays}; + new_stra_arrays[i][j] = new_stra_arrays[i][j] * dev_num_not_2_power; + if (i == 0 && j == related_dim_left) { + new_stra_arrays[1][related_dim_right] = new_stra_arrays[1][related_dim_right] * dev_num_not_2_power; + } + StrategyPtr new_stra = std::make_shared(stage_id, new_stra_arrays); + sp_vector.push_back(new_stra); + } + } + } + strategy_cost_.clear(); + for (auto &sp : sp_vector) { + if (SetCostUnderStrategy(sp) == FAILED) { + MS_LOG(WARNING) << name_ << " : Calculating cost for strategy failed."; + continue; + } + } + return SUCCESS; +} + Status MatMulBase::GenerateStrategies(int64_t stage_id) { if (GetAttrs() != SUCCESS) { MS_LOG(ERROR) << name_ << " : GetAttrs failed."; @@ -424,54 +510,28 @@ Status MatMulBase::GenerateStrategies(int64_t stage_id) { MS_LOG(ERROR) << name_ << " : Swap last two elements failed."; } } - // The shape of input0 (input1) - // E.g., input0 = [100, 200, 300], input1 = [300, 400] - - // Combining the input0_shape and input1_shape - // E.g., combined_shape = [100, 200, 300, 400] - size_t input1_shape_size = input1_shape.size(), input0_shape_size = input0_shape.size(); - Dimensions combined_partitions; - Shape combined_shape; - // In SwapLastTwoElements(), it is guaranteed that input0_shape.size() and input1_shape.size() are both larger than 2 - if (input0_shape.size() >= input1_shape.size()) { - combined_shape = input0_shape; - combined_shape.push_back(input1_shape[input1_shape.size() - 1]); - } else { - combined_shape = input1_shape; - combined_shape.push_back(input0_shape[input0_shape.size() - 2]); - } - std::function recursive = [&stage_id, &dev_num, &combined_partitions, &combined_shape, - &input1_shape_size, &recursive, &input0_shape_size, - this](uint64_t current_index, size_t n) { - // Finishing the recursive steps, if the strategy is valid, then calculate the cost - // for this operator under the strategy. - if (current_index == combined_shape.size()) { - StrategyPtr sp; - if (this->PrepareStrategy(stage_id, dev_num, combined_partitions, input0_shape_size, input1_shape_size, &sp) == - FAILED) { - return; - } - if (this->SetCostUnderStrategy(sp) == FAILED) { + auto dev_num_2_power = (dev_num & (dev_num - 1)); + std::vector sp_vector_2_power_part; + if (dev_num_2_power == 0) { + if (GenerateStrategiesBase(stage_id, dev_num, input0_shape, input1_shape, &sp_vector_2_power_part) != SUCCESS) { + return FAILED; + } + strategy_cost_.clear(); + for (auto &sp : sp_vector_2_power_part) { + if (SetCostUnderStrategy(sp) == FAILED) { MS_LOG(WARNING) << name_ << " : Calculating cost for strategy failed."; - return; - } - } else { - MS_LOG(DEBUG) << name_ << " : The value input0_shape_size: " << input0_shape_size - << ", input1_shape_size: " << input1_shape_size; - for (uint64_t i = 1; i <= n; i *= 2) { - if (n % i == 0 && LongToSize(combined_shape[current_index]) % i == 0) { - combined_partitions.push_back(i); - recursive(current_index + 1, n / i); - combined_partitions.pop_back(); - } + continue; } } - }; - recursive(0, dev_num); - if (strategy_cost_.empty()) { - MS_LOG(EXCEPTION) << name_ << " : No available strategy."; + return SUCCESS; } - return Status::SUCCESS; + auto dev_num_not_2_power = dev_num / (dev_num - dev_num_2_power); + if (GenerateStrategiesBase(stage_id, dev_num - dev_num_2_power, input0_shape, input1_shape, + &sp_vector_2_power_part) != SUCCESS) { + MS_LOG(ERROR) << "Generating strategy in power of 2 devices failed."; + return FAILED; + } + return GenerateStrategiesNotPower2(stage_id, dev_num_not_2_power, sp_vector_2_power_part); } std::vector MatMulBase::GenerateOpStrategies(int64_t) { @@ -601,7 +661,7 @@ std::shared_ptr BatchMatMulInfo::GenerateBatchStrategies() { Status MatMulBase::SetCostUnderStrategy(const mindspore::parallel::StrategyPtr &strategy) { if (InitForCostModel(strategy, nullptr) == FAILED) { - MS_LOG(ERROR) << name_ << " : Initialization under the strategy failed."; + MS_LOG(INFO) << name_ << " : Initialization under the strategy failed."; return FAILED; } PrintStrategy(strategy); diff --git a/mindspore/ccsrc/frontend/parallel/ops_info/matmul_info.h b/mindspore/ccsrc/frontend/parallel/ops_info/matmul_info.h index 33221cae587..a00c8aaf332 100644 --- a/mindspore/ccsrc/frontend/parallel/ops_info/matmul_info.h +++ b/mindspore/ccsrc/frontend/parallel/ops_info/matmul_info.h @@ -53,6 +53,10 @@ class MatMulBase : public OperatorInfo { Status InferTensorMap() override; Status InferTensorLayout(TensorLayouts *inputs_layout, TensorLayouts *outputs_layout); void InitTensorInfoForCost(std::vector *); + Status GenerateStrategiesBase(int64_t stage_id, size_t dev_num, const Shape &input0_shape, Shape input1_shape, + std::vector *const sp_vector); + Status GenerateStrategiesNotPower2(int64_t stage_id, size_t dev_num_not_2_power, + const std::vector &sp_vector_2_power_part); Status CheckForTensorSliceValid() const; Status GetAttrs() override; diff --git a/mindspore/ccsrc/frontend/parallel/ops_info/operator_info.cc b/mindspore/ccsrc/frontend/parallel/ops_info/operator_info.cc index 30c9331faa5..a19cb3e3b85 100644 --- a/mindspore/ccsrc/frontend/parallel/ops_info/operator_info.cc +++ b/mindspore/ccsrc/frontend/parallel/ops_info/operator_info.cc @@ -187,7 +187,7 @@ Status OperatorInfo::InferMirrorOps() { for (size_t i = 0; i < inputs_tensor_map_.size(); ++i) { std::vector group; if (CreateGroupByTensorMap(inputs_tensor_map_[i], &group) != SUCCESS) { - MS_LOG(ERROR) << name_ << ": Create group failed, the input index is " << i; + ReportError(name_ + ": Create group failed, the input index is " + std::to_string(i)); mirror_ops_.clear(); return FAILED; } @@ -590,8 +590,24 @@ Status OperatorInfo::CreateGroupByTensorMap(const Shape &tensor_map, std::vector MS_LOG(INFO) << "The dev size is 1, no need to create group."; return SUCCESS; } + if (is_auto_parallel_) { + if (g_device_manager->CheckDeviceList(group_devices) != SUCCESS) { + MS_LOG(INFO) << "Try to create communication group : " << group_devices + << " failed in auto parallel mode, " + "this error can be ignored in parallel strategies searching step"; + return FAILED; + } + return SUCCESS; + } - Group g = g_device_manager->CreateGroup(group_devices); + Group g; + if (g_device_manager->CreateGroup(group_devices, &g) != SUCCESS) { + MS_LOG(ERROR) << "Operator " << name_ + << " create communication group by tensor_map failed, the rank_list is: " << group_devices + << ", the input strategy is " << strategy_->GetInputDim() + << ", the full_name of node is: " << cnode_->fullname_with_scope(); + return FAILED; + } group->push_back(g); return SUCCESS; } @@ -630,7 +646,15 @@ Status OperatorInfo::CreateGroupForOptShard(TensorLayout *const tensor_layout, s RankList new_group_devices( group_devices.begin() + index / optimizer_weight_shard_size * optimizer_weight_shard_size, group_devices.begin() + (index / optimizer_weight_shard_size + 1) * optimizer_weight_shard_size); - Group allgather_group = g_device_manager->CreateGroup(new_group_devices); + Group allgather_group; + if (g_device_manager->CreateGroup(new_group_devices, &allgather_group) != SUCCESS) { + MS_LOG(ERROR) << "Operator " << name_ + << " create communication group for allgather in optimizer parallel failed," + " the rank_list is: " + << group_devices << ", the input strategy is " << strategy_->GetInputDim() + << ", the full_name of node is: " << cnode_->fullname_with_scope(); + return FAILED; + } groups->push_back(allgather_group); tensor_layout->set_opt_shard_group(allgather_group.name()); MS_LOG(INFO) << "Parallel optimizer: create allgather group " << allgather_group.name(); @@ -643,14 +667,30 @@ Status OperatorInfo::CreateGroupForOptShard(TensorLayout *const tensor_layout, s if (temp_dev_matrix.GetDevicesAlongDim(0, &mirror_group_devices) != SUCCESS) { return FAILED; } - Group mirror_group = g_device_manager->CreateGroup(mirror_group_devices); + Group mirror_group; + if (g_device_manager->CreateGroup(mirror_group_devices, &mirror_group)) { + MS_LOG(ERROR) << "Operator " << name_ + << " create communication group for mirror in optimizer parallel failed," + " the rank_list is: " + << group_devices << ", the input strategy is " << strategy_->GetInputDim() + << ", the full_name of node is: " << cnode_->fullname_with_scope(); + return FAILED; + } groups->push_back(mirror_group); tensor_layout->set_opt_shard_mirror_group(mirror_group.name()); MS_LOG(INFO) << "Parallel optimizer: create mirror group " << mirror_group.name(); } else { // fully use opt shard // create allgather group - Group allgather_group = g_device_manager->CreateGroup(group_devices); + Group allgather_group; + if (g_device_manager->CreateGroup(group_devices, &allgather_group) != SUCCESS) { + MS_LOG(ERROR) << "Operator " << name_ + << " create communication group for allgather in optimizer parallel failed," + " the rank_list is: " + << group_devices << ", the input strategy is " << strategy_->GetInputDim() + << ", the full_name of node is: " << cnode_->fullname_with_scope(); + return FAILED; + } groups->push_back(allgather_group); tensor_layout->set_opt_shard_group(allgather_group.name()); MS_LOG(INFO) << "Parallel optimizer: create allgather group " << allgather_group.name(); @@ -684,8 +724,23 @@ Status OperatorInfo::CreateGroupByDim(size_t axis, std::vector *group) { MS_LOG(INFO) << "The dev size is 1, no need to create group."; return SUCCESS; } - - Group g = g_device_manager->CreateGroup(group_devices); + if (is_auto_parallel_) { + if (g_device_manager->CheckDeviceList(group_devices) != SUCCESS) { + MS_LOG(INFO) << "Try to create communication group : " << group_devices + << " failed in auto parallel mode, " + "this error can be ignored in parallel strategies searching step"; + return FAILED; + } + return SUCCESS; + } + Group g; + if (g_device_manager->CreateGroup(group_devices, &g) != SUCCESS) { + MS_LOG(ERROR) << "Operator " << name_ + << " create communication group by dim failed, the rank_list is: " << group_devices + << ", the input strategy is " << strategy_->GetInputDim() + << ", the full_name of node is: " << cnode_->fullname_with_scope(); + return FAILED; + } group->push_back(g); return SUCCESS; } @@ -774,7 +829,7 @@ Status OperatorInfo::Init(const StrategyPtr &in_strategy, const StrategyPtr &out Status OperatorInfo::InitForCostModel(const StrategyPtr &in_strategy, const StrategyPtr &out_strategy) { if (InitForCostModelWithAutoRepeatCalc(in_strategy, out_strategy) != SUCCESS) { - MS_LOG(ERROR) << name_ << " : Init for cost model failed."; + ReportError(name_ + " : Init for cost model failed."); return FAILED; } @@ -844,7 +899,19 @@ Status OperatorInfo::InitForCostModelWithAutoRepeatCalc(const StrategyPtr &in_st MS_LOG(ERROR) << name_ << ": InferTensorInfo failed."; return FAILED; } + auto stage_dev_num = g_device_manager->stage_device_num(); + if ((stage_dev_num & (stage_dev_num - 1)) == 0) { + return SUCCESS; + } + if (InferForwardCommunication() != SUCCESS) { + MS_LOG(WARNING) << name_ << ": InferForwardCommunication failed in auto parallel searching strategies step."; + return FAILED; + } + if (InferMirrorOps() != SUCCESS) { + MS_LOG(WARNING) << name_ << ": InferMirrorOps failed in auto parallel searching strategies step."; + return FAILED; + } return SUCCESS; } @@ -1244,26 +1311,9 @@ Status GenerateStrategiesForBroadcastBoth(int64_t stage_id, const Shapes &inputs return SUCCESS; } -// 'splittable_inputs' has the same dimensions as 'inputs_shape_'. '0' in 'splittable_inputs' means that -// the corresponding dimension is unsplittable, '1' in 'splittable_inputs' means that the corresponding -// dimension is splittable. 'inputs_partitions' is the result of partitions. -// NOTE: This implementation would partition all splittable dimensions in all inputs. Some operators requiring -// specific dimensions in inputs have the identical partition should have individual implementation. -Status GenerateStrategiesForIndependentInputs(int64_t stage_id, const Shapes &inputs_shape, - const Shapes &splittable_inputs, - std::vector *const sp_vector) { - if (sp_vector == nullptr) { - MS_LOG(ERROR) << "The sp_vector is null."; - return FAILED; - } - if (splittable_inputs.size() != inputs_shape.size()) { - MS_LOG(ERROR) << "Splittable_inputs do not have the same input number of inputs shape, " << splittable_inputs.size() - << " : " << inputs_shape.size(); - return FAILED; - } - CheckGlobalDeviceManager(); - size_t dev_num = g_device_manager->GetDeviceListByStageId(stage_id).size(); - +Status GenerateStrategiesForIndependentInputsBase(int64_t stage_id, size_t dev_num, const Shapes &inputs_shape, + const Shapes &splittable_inputs, + std::vector *const sp_vector) { Shape combined_inputs_shape, combined_splittable_inputs, combined_partitions; for (size_t j = 0; j < inputs_shape.size(); ++j) { (void)combined_inputs_shape.insert(combined_inputs_shape.end(), inputs_shape[j].begin(), inputs_shape[j].end()); @@ -1314,6 +1364,57 @@ Status GenerateStrategiesForIndependentInputs(int64_t stage_id, const Shapes &in return SUCCESS; } +// 'splittable_inputs' has the same dimensions as 'inputs_shape_'. '0' in 'splittable_inputs' means that +// the corresponding dimension is unsplittable, '1' in 'splittable_inputs' means that the corresponding +// dimension is splittable. 'inputs_partitions' is the result of partitions. +// NOTE: This implementation would partition all splittable dimensions in all inputs. Some operators requiring +// specific dimensions in inputs have the identical partition should have individual implementation. +Status GenerateStrategiesForIndependentInputs(int64_t stage_id, const Shapes &inputs_shape, + const Shapes &splittable_inputs, + std::vector *const sp_vector) { + if (sp_vector == nullptr) { + MS_LOG(ERROR) << "The sp_vector is null."; + return FAILED; + } + if (splittable_inputs.size() != inputs_shape.size()) { + MS_LOG(ERROR) << "Splittable_inputs do not have the same input number of inputs shape, " << splittable_inputs.size() + << " : " << inputs_shape.size(); + return FAILED; + } + CheckGlobalDeviceManager(); + size_t dev_num = g_device_manager->GetDeviceListByStageId(stage_id).size(); + auto dev_num_2_power = (dev_num & (dev_num - 1)); + if (dev_num_2_power == 0) { + return GenerateStrategiesForIndependentInputsBase(stage_id, dev_num, inputs_shape, splittable_inputs, sp_vector); + } + auto dev_num_not_2_power = dev_num / (dev_num - dev_num_2_power); + std::vector sp_vector_2_power_part; + if (GenerateStrategiesForIndependentInputsBase(stage_id, dev_num - dev_num_2_power, inputs_shape, splittable_inputs, + &sp_vector_2_power_part) != SUCCESS) { + MS_LOG(ERROR) << "Generate strategy in the power of 2 devices part failed."; + return FAILED; + } + // Handle the not power of 2 part. + for (auto &stra : sp_vector_2_power_part) { + auto stra_arrays = stra->GetInputDim(); + size_t stras_size = stra_arrays.size(); + for (size_t i = 0; i < stras_size; ++i) { + auto split_input = splittable_inputs[i]; + size_t stra_size = stra_arrays[i].size(); + for (size_t j = 0; j < stra_size; ++j) { + if (split_input[j] == 0) { + continue; + } + auto new_stra_arrays{stra_arrays}; + new_stra_arrays[i][j] = new_stra_arrays[i][j] * dev_num_not_2_power; + StrategyPtr new_stra = std::make_shared(stage_id, new_stra_arrays); + sp_vector->push_back(new_stra); + } + } + } + return SUCCESS; +} + // generate strategies for that have two inputs, and input0 or input1 maybe broadcast, // and the corresponding dimensions that are not broadcast are all relevant dimensions // such as: ([a, b, c, d], [a, b, c, d]) or ([b, c, d], [a, b, c, d]) or ([1, c, d], [a, b, c, d]) diff --git a/mindspore/ccsrc/frontend/parallel/ops_info/operator_info.h b/mindspore/ccsrc/frontend/parallel/ops_info/operator_info.h index 7d2254eee00..2dcfafb3e31 100644 --- a/mindspore/ccsrc/frontend/parallel/ops_info/operator_info.h +++ b/mindspore/ccsrc/frontend/parallel/ops_info/operator_info.h @@ -165,7 +165,10 @@ class OperatorInfo { void SetIsStrategyCostExactTrue() { is_strategy_cost_exact_ = true; } void ClearStrategyCost() { strategy_cost_.clear(); } void CheckSelectedStrategy(const StrategyPtr &); - Status InitSelectedStrategy(const StrategyPtr &s_strategy) { return Init(s_strategy, nullptr); } + Status InitSelectedStrategy(const StrategyPtr &s_strategy) { + set_auto_parallel(false); + return Init(s_strategy, nullptr); + } void set_input_value(const std::vector &input_value) { input_value_ = input_value; } const std::vector &input_value() const { return input_value_; } void set_outputs_dtype(const TypePtr &dtype) { outputs_dtype_ = dtype; } @@ -201,6 +204,7 @@ class OperatorInfo { Status CreateGroupByTensorMap(const Shape &tensor_map, std::vector *group); Status CreateGroupForOptShard(TensorLayout *const tensor_layout, std::vector *group); virtual void ReplaceNodeInputOrAttrs() {} + void set_auto_parallel(bool is_auto_parallel) { is_auto_parallel_ = is_auto_parallel; } // Key for user data. constexpr static char key[] = "OpInfo"; @@ -241,6 +245,13 @@ class OperatorInfo { float GetFloatAttr(const std::string &attr_name); std::string GetStringAttr(const std::string &attr_name); std::vector GetTupleIntAttr(const std::string &attr_name); + void ReportError(const std::string &error_msg) { + if (is_auto_parallel_) { + MS_LOG(INFO) << error_msg; + } else { + MS_LOG(ERROR) << error_msg; + } + } std::string name_; Shapes inputs_shape_; @@ -339,6 +350,9 @@ std::shared_ptr GenerateBatchStrategiesBySplitFlag(const Shapes &shap const std::vector &split_flag_list); std::string StrategyToString(const Strategys &strategy); void PrintStrategy(const StrategyPtr &strategy); +Status GenerateStrategiesForIndependentInputsBase(int64_t stage_id, size_t dev_num, const Shapes &inputs_shape, + const Shapes &splittable_inputs, + std::vector *const sp_vector); // generate strategies for that all inputs' dimensions are independent, such as: ([a, b, c, d]) Status GenerateStrategiesForIndependentInputs(int64_t stage_id, const Shapes &inputs_shape, const Shapes &splittable_inputs, std::vector *sp_vector); diff --git a/mindspore/ccsrc/frontend/parallel/ops_info/reduce_method_info.cc b/mindspore/ccsrc/frontend/parallel/ops_info/reduce_method_info.cc index 67cefd7fcf6..175ab6e1a08 100644 --- a/mindspore/ccsrc/frontend/parallel/ops_info/reduce_method_info.cc +++ b/mindspore/ccsrc/frontend/parallel/ops_info/reduce_method_info.cc @@ -186,7 +186,7 @@ Status ReduceMethod::InferForwardCommunication() { } std::vector forward_group; if (CreateGroupByTensorMap(group_creat_map, &forward_group) != SUCCESS) { - MS_LOG(ERROR) << name_ << ": InferForwardCommunication group failed."; + ReportError(name_ + ": Create group failed."); return FAILED; } if (!forward_group.empty()) { @@ -264,7 +264,7 @@ Status ReduceMeanInfo::InferForwardCommunication() { std::vector forward_group; if (CreateGroupByTensorMap(group_creat_map, &forward_group) != SUCCESS) { - MS_LOG(ERROR) << name_ << ": InferForwardCommunication group failed."; + ReportError(name_ + ": Create group failed."); return FAILED; } if (!forward_group.empty()) { @@ -285,7 +285,7 @@ Status ReduceMethod::InferMirrorOps() { Shape input_tensor_map = inputs_tensor_map_.at(0); std::vector input_group; if (CreateGroupByTensorMap(input_tensor_map, &input_group) != SUCCESS) { - MS_LOG(ERROR) << name_ << " Infer MirrorOps failed."; + ReportError(name_ + ": Create group failed."); return FAILED; } @@ -310,7 +310,7 @@ Status ArgMaxWithValueInfo::InferMirrorOps() { Shape input_tensor_map = inputs_tensor_map_.at(0); std::vector input_group; if (CreateGroupByTensorMap(input_tensor_map, &input_group) != SUCCESS) { - MS_LOG(ERROR) << name_ << ": Infer MirrorOps failed."; + ReportError(name_ + ": Create group failed."); return FAILED; } diff --git a/mindspore/ccsrc/frontend/parallel/ops_info/reshape_info.cc b/mindspore/ccsrc/frontend/parallel/ops_info/reshape_info.cc index 68c4fb475dd..5de78e6129b 100644 --- a/mindspore/ccsrc/frontend/parallel/ops_info/reshape_info.cc +++ b/mindspore/ccsrc/frontend/parallel/ops_info/reshape_info.cc @@ -51,7 +51,7 @@ Status ReshapeInfo::InferMirrorOps() { Shape input_tensor_map = input_layout_.tensor_map().array(); std::vector input_group; if (CreateGroupByTensorMap(input_tensor_map, &input_group) != SUCCESS) { - MS_LOG(ERROR) << name_ << ": Infer MirrorOps failed."; + ReportError(name_ + ": Create group failed."); return FAILED; } diff --git a/mindspore/ccsrc/frontend/parallel/ops_info/roi_align_info.cc b/mindspore/ccsrc/frontend/parallel/ops_info/roi_align_info.cc index 86e56690e03..bfaa926f7fb 100644 --- a/mindspore/ccsrc/frontend/parallel/ops_info/roi_align_info.cc +++ b/mindspore/ccsrc/frontend/parallel/ops_info/roi_align_info.cc @@ -126,7 +126,9 @@ Status ROIAlignInfo::InferGroup() { } MS_LOG(INFO) << name_ << ": The group rank is " << group_devices; - group_ = g_device_manager->CreateGroup(group_devices); + if (g_device_manager->CreateGroup(group_devices, &group_) != SUCCESS) { + MS_LOG(ERROR) << "The node " << cnode_->fullname_with_scope() << " create sync allreduce failed"; + } return SUCCESS; } diff --git a/mindspore/ccsrc/frontend/parallel/ops_info/slice_info.cc b/mindspore/ccsrc/frontend/parallel/ops_info/slice_info.cc index d2d73c6871f..ea7d0415d7f 100644 --- a/mindspore/ccsrc/frontend/parallel/ops_info/slice_info.cc +++ b/mindspore/ccsrc/frontend/parallel/ops_info/slice_info.cc @@ -131,7 +131,7 @@ Status SliceInfo::InferMirrorOps() { Shape input_tensor_map = inputs_tensor_map_[0]; std::vector group; if (CreateGroupByTensorMap(input_tensor_map, &group) != SUCCESS) { - MS_LOG(ERROR) << name_ << ": Create group for input failed."; + ReportError(name_ + ": Create group failed."); return FAILED; } diff --git a/mindspore/ccsrc/frontend/parallel/ops_info/strided_slice_info.cc b/mindspore/ccsrc/frontend/parallel/ops_info/strided_slice_info.cc index 11d118e30bd..b0679ef9bfc 100644 --- a/mindspore/ccsrc/frontend/parallel/ops_info/strided_slice_info.cc +++ b/mindspore/ccsrc/frontend/parallel/ops_info/strided_slice_info.cc @@ -166,7 +166,7 @@ Status StridedSliceInfo::InferMirrorOps() { Shape input_tensor_map = inputs_tensor_map_[0]; std::vector group; if (CreateGroupByTensorMap(input_tensor_map, &group) != SUCCESS) { - MS_LOG(ERROR) << name_ << ": Create group for input failed."; + ReportError(name_ + ": Create group failed."); return FAILED; } diff --git a/mindspore/ccsrc/frontend/parallel/ops_info/tensordot_info.cc b/mindspore/ccsrc/frontend/parallel/ops_info/tensordot_info.cc index 937e17db558..2a6cd34997d 100644 --- a/mindspore/ccsrc/frontend/parallel/ops_info/tensordot_info.cc +++ b/mindspore/ccsrc/frontend/parallel/ops_info/tensordot_info.cc @@ -196,7 +196,7 @@ Status TensorDotInfo::InferForwardCommunication() { std::vector forward_group; if (CreateGroupByTensorMap(forward_group_map, &forward_group) != SUCCESS) { - MS_LOG(ERROR) << name_ << ": Create group by tensor map failed"; + ReportError(name_ + ": Create group failed."); return FAILED; } diff --git a/mindspore/ccsrc/frontend/parallel/ops_info/tile_info.cc b/mindspore/ccsrc/frontend/parallel/ops_info/tile_info.cc index db1a29d50b2..b495f8c10cc 100644 --- a/mindspore/ccsrc/frontend/parallel/ops_info/tile_info.cc +++ b/mindspore/ccsrc/frontend/parallel/ops_info/tile_info.cc @@ -139,7 +139,7 @@ Status TileInfo::InferMirrorOps() { Shape input_tensor_map = inputs_tensor_map_[0]; std::vector group; if (CreateGroupByTensorMap(input_tensor_map, &group) != SUCCESS) { - MS_LOG(ERROR) << name_ << ": Create group for input failed."; + ReportError(name_ + ": Create group failed."); return FAILED; } diff --git a/mindspore/ccsrc/frontend/parallel/ops_info/topk_info.cc b/mindspore/ccsrc/frontend/parallel/ops_info/topk_info.cc index 4aa80dab406..efdab434087 100644 --- a/mindspore/ccsrc/frontend/parallel/ops_info/topk_info.cc +++ b/mindspore/ccsrc/frontend/parallel/ops_info/topk_info.cc @@ -120,7 +120,7 @@ Status TopKInfo::InferMirrorOps() { for (size_t i = 0; i < inputs_tensor_map_.size(); ++i) { std::vector group; if (CreateGroupByTensorMap(inputs_tensor_map_[i], &group) != SUCCESS) { - MS_LOG(ERROR) << name_ << ": Create group failed, the input index is " << i; + ReportError(name_ + ": Create group failed, the input index is " + std::to_string(i)); mirror_ops_.clear(); return FAILED; } diff --git a/mindspore/ccsrc/frontend/parallel/ops_info/unsorted_segment_op_info.cc b/mindspore/ccsrc/frontend/parallel/ops_info/unsorted_segment_op_info.cc index 27fbb8f6116..cb78e72e2cf 100644 --- a/mindspore/ccsrc/frontend/parallel/ops_info/unsorted_segment_op_info.cc +++ b/mindspore/ccsrc/frontend/parallel/ops_info/unsorted_segment_op_info.cc @@ -104,7 +104,7 @@ Status UnsortedSegmentOpInfo::InferMirrorOps() { Shape tensor_map = inputs_tensor_map_[0]; std::vector group; if (CreateGroupByTensorMap(tensor_map, &group) != SUCCESS) { - MS_LOG(ERROR) << name_ << " : Create group failed."; + ReportError(name_ + ": Create group failed."); return FAILED; } @@ -198,7 +198,7 @@ Status UnsortedSegmentOpInfo::InferForwardCommunication() { tmp_group_tensor_map.push_back(0); } if (CreateGroupByTensorMap(tmp_group_tensor_map, &group_list) != SUCCESS) { - MS_LOG(ERROR) << name_ << " : Infer forward communication, create group failed."; + ReportError(name_ + ": Create group failed."); return FAILED; } else if (group_list.empty()) { MS_LOG(INFO) << name_ << " : Forward all reduce is not required."; diff --git a/mindspore/ccsrc/frontend/parallel/parameter_manager.cc b/mindspore/ccsrc/frontend/parallel/parameter_manager.cc index 1780e843d43..4c2af46cf55 100644 --- a/mindspore/ccsrc/frontend/parallel/parameter_manager.cc +++ b/mindspore/ccsrc/frontend/parallel/parameter_manager.cc @@ -764,7 +764,10 @@ std::vector IsBorderAdaSumSendReceive(const AnfNodePtr &node, const RankLi group_list = {rank, origin_dest_rank}; new_dest_src_rank = 1; } - Group adasum_send_rec_group = g_device_manager->CreateGroup(group_list); + Group adasum_send_rec_group; + if (g_device_manager->CreateGroup(group_list, &adasum_send_rec_group) != SUCCESS) { + MS_LOG(EXCEPTION) << "Create send/receive group in adasum failed, the group is:" << group_list; + } send_rec_prim->set_attr(GROUP, MakeValue(adasum_send_rec_group.name())); if (is_send) { send_rec_prim->set_attr(DEST_RANK, MakeValue(new_dest_src_rank)); @@ -843,7 +846,10 @@ void HandleAdasumAllReduce(const PrimitivePtr &prim, const RankList &group_devic int64_t neighbor_id = (node_rank / double_d * double_d + index) * ADASUM_MIN_DIS + rank % ADASUM_MIN_DIS; neighbor_ids.push_back(neighbor_id); } - Group adasum_allreduce_group = g_device_manager->CreateGroup(neighbor_ids); + Group adasum_allreduce_group; + if (g_device_manager->CreateGroup(neighbor_ids, &adasum_allreduce_group) != SUCCESS) { + MS_LOG(EXCEPTION) << "Create group allreduce group in adasum failed, the group is " << neighbor_ids; + } auto new_group_name = MakeValue(adasum_allreduce_group.name()); int64_t fusion_id = GetValue(prim->GetAttr("origin_fusion")); int64_t new_fusion_id = fusion_id + g_device_manager->DeviceNum() * (border_step + 1); @@ -1018,7 +1024,10 @@ void ResetMirrorAttr(const PrimitivePtr &prim, const RankList &new_group) { prim->set_attr(GROUP_RANKS, MakeValue(std::to_string(new_group[0]))); return; } - Group adasum_mirror_group = g_device_manager->CreateGroup(new_group); + Group adasum_mirror_group; + if (g_device_manager->CreateGroup(new_group, &adasum_mirror_group) != SUCCESS) { + MS_LOG(EXCEPTION) << "Create new mirror group failed in adasum, new group is: " << new_group; + } auto new_group_name = MakeValue(adasum_mirror_group.name()); prim->set_attr(GROUP, new_group_name); prim->set_attr(DEV_NUM, MakeValue(new_group.size())); diff --git a/mindspore/ccsrc/frontend/parallel/pipeline_transformer/pipeline_transformer.cc b/mindspore/ccsrc/frontend/parallel/pipeline_transformer/pipeline_transformer.cc index 71e02d25bcf..81e63eeadaa 100644 --- a/mindspore/ccsrc/frontend/parallel/pipeline_transformer/pipeline_transformer.cc +++ b/mindspore/ccsrc/frontend/parallel/pipeline_transformer/pipeline_transformer.cc @@ -181,9 +181,17 @@ void PipelineTransformer::LabelMicroBatch() { void PipelineTransformer::CreateForwardGroup() { std::vector rank_list = g_device_manager->GetDeviceListBetweenStage(); auto dev_list = g_device_manager->CreateDeviceListByRankList(rank_list); - auto g = g_device_manager->CreateGroup(rank_list); + Group g; + if (g_device_manager->CreateGroup(rank_list, &g) != SUCCESS) { + MS_LOG(EXCEPTION) << "Create forward communication group between all pipeline stages failed, the rank_list is: " + << rank_list; + } auto g_back_name = g.name() + BACKWARD; - auto g_back = g_device_manager->CreateGroup(g_back_name, dev_list); + Group g_back; + if (g_device_manager->CreateGroup(g_back_name, dev_list, &g_back) != SUCCESS) { + MS_LOG(EXCEPTION) << "Create backward communication group between all pipeline stages failed, the rank_list is: " + << rank_list; + } group_.push_back(g.name()); group_.push_back(g_back.name()); } diff --git a/mindspore/ccsrc/frontend/parallel/step_auto_parallel.cc b/mindspore/ccsrc/frontend/parallel/step_auto_parallel.cc index ff63f1ba0cf..f17ff828153 100644 --- a/mindspore/ccsrc/frontend/parallel/step_auto_parallel.cc +++ b/mindspore/ccsrc/frontend/parallel/step_auto_parallel.cc @@ -90,6 +90,11 @@ bool StepAutoParallel(const FuncGraphPtr &root, const opt::OptimizerPtr &) { if (ParallelInit() != SUCCESS) { MS_LOG(EXCEPTION) << "Parallel init failed"; } + if (strategy_search_mode == kRecursiveProgramming && + ((g_device_manager->DeviceNum() & (g_device_manager->DeviceNum() - 1)) != 0)) { + MS_LOG(EXCEPTION) + << "The recursive auto parallel strategy searching mode requires the device num be the power of 2."; + } // mark the forward cnodes, parallel only care these nodes MarkForwardCNode(root); if (IsInsertVirtualOutput(root)) { @@ -354,6 +359,7 @@ OperatorInfoPtr CreateTheOperatorInfo(const PrimitivePtr &prim, const CNodePtr & operator_info->set_input_value(input_value); operator_info->set_outputs_dtype(cnode->Type()); operator_info->set_cnode(cnode); + operator_info->set_auto_parallel(true); AddOperatorToIgnoreCandidates(prim, operator_info); // key of strategy map diff --git a/mindspore/ccsrc/frontend/parallel/step_parallel.cc b/mindspore/ccsrc/frontend/parallel/step_parallel.cc index c267f840af5..2df377fcc2b 100644 --- a/mindspore/ccsrc/frontend/parallel/step_parallel.cc +++ b/mindspore/ccsrc/frontend/parallel/step_parallel.cc @@ -3178,13 +3178,23 @@ static void InsertAllReduceForNormValue(const AnfNodePtr &res_node) { auto sqrt_node = MatchPattern(expand_dims_node, node_user_map, REDUCE_SUM_MATCH_PATTERN); if (!sqrt_node) return; auto cur_stage_rank_list = g_device_manager->GetDeviceListInThisStage(); - Group cur_stage_device_list = g_device_manager->CreateGroup(cur_stage_rank_list); + Group cur_stage_device_list; + if (g_device_manager->CreateGroup(cur_stage_rank_list, &cur_stage_device_list) != SUCCESS) { + MS_LOG(EXCEPTION) << "Create the communication group for allreduce in calculating global norm failed, " + "the rank_list is: " + << cur_stage_rank_list; + } InsertAllReduceToNodeInput(sqrt_node->cast(), cur_stage_device_list.name(), PARALLEL_GLOBALNORM); MS_LOG(INFO) << "Insert the AllReduce for global norm value in stages succeed."; if (pipeline_stages > 1) { MS_LOG(INFO) << "Insert the AllReduce for global norm value between stages succeed."; auto ranks_between_stages = g_device_manager->GetDeviceListBetweenStage(); - Group group_between_stages = g_device_manager->CreateGroup(ranks_between_stages); + Group group_between_stages; + if (g_device_manager->CreateGroup(ranks_between_stages, &group_between_stages)) { + MS_LOG(EXCEPTION) << "Create the communication group for allreduce in calculating global norm " + "with pipeline parallel failed, the rank_list is: " + << cur_stage_rank_list; + } InsertAllReduceToNodeInput(sqrt_node->cast(), group_between_stages.name(), PARALLEL_GLOBALNORM_BETWEEN); } } diff --git a/mindspore/ccsrc/frontend/parallel/tensor_layout/arrangement.cc b/mindspore/ccsrc/frontend/parallel/tensor_layout/arrangement.cc index 6e1ef8ef503..c1e78b9c1b5 100644 --- a/mindspore/ccsrc/frontend/parallel/tensor_layout/arrangement.cc +++ b/mindspore/ccsrc/frontend/parallel/tensor_layout/arrangement.cc @@ -115,7 +115,7 @@ std::shared_ptr> Arrangement::GetExpandShapeList(const for (size_t i = 0; i < expand_shape.GetDimSize(); i++) { size *= expand_shape.GetDimByIdx(i); if (size > GetDimByIdx(ind)) { - MS_LOG(ERROR) << "invalid expand_shape"; + MS_LOG(INFO) << "invalid expand_shape:" << expand_shape.array(); return nullptr; } else if (size < GetDimByIdx(ind)) { shape.push_back(expand_shape.GetDimByIdx(i)); @@ -131,7 +131,7 @@ std::shared_ptr> Arrangement::GetExpandShapeList(const } } if (ind != GetDimSize()) { - MS_LOG(ERROR) << "invalid expand_shape"; + MS_LOG(INFO) << "invalid expand_shape:" << expand_shape.array(); return nullptr; } auto arrangement_new = std::make_shared>(arrangement_list); diff --git a/mindspore/ccsrc/frontend/parallel/tensor_layout/construct_operator.cc b/mindspore/ccsrc/frontend/parallel/tensor_layout/construct_operator.cc index 5ce022eea51..c928e6179cf 100644 --- a/mindspore/ccsrc/frontend/parallel/tensor_layout/construct_operator.cc +++ b/mindspore/ccsrc/frontend/parallel/tensor_layout/construct_operator.cc @@ -22,10 +22,11 @@ namespace mindspore { namespace parallel { -Status ConstructOperator::Init(const RankList &dev_list, const Shape &dev_matrix_shape) { +Status ConstructOperator::Init(const RankList &dev_list, const Shape &dev_matrix_shape, bool is_cost_model) { dev_size_ = dev_matrix_shape.size(); dev_matrix_shape_ = dev_matrix_shape; dev_list_ = dev_list; + is_cost_model_ = is_cost_model; return Status::SUCCESS; } @@ -260,8 +261,14 @@ Status ConstructOperator::CreateGroupByDim(size_t axis, std::vector *grou MS_LOG(INFO) << "the group is empty"; return SUCCESS; } - - Group g = g_device_manager->CreateGroup(group_devices); + if (is_cost_model_) { + return g_device_manager->CheckDeviceList(group_devices); + } + Group g; + if (g_device_manager->CreateGroup(group_devices, &g) != SUCCESS) { + MS_LOG(ERROR) << "Create communication group in redistribution failed, the rank_list is: " << group_devices; + return FAILED; + } group->push_back(g); return SUCCESS; } diff --git a/mindspore/ccsrc/frontend/parallel/tensor_layout/construct_operator.h b/mindspore/ccsrc/frontend/parallel/tensor_layout/construct_operator.h index 9d913c02d58..0e691cdca8e 100644 --- a/mindspore/ccsrc/frontend/parallel/tensor_layout/construct_operator.h +++ b/mindspore/ccsrc/frontend/parallel/tensor_layout/construct_operator.h @@ -34,7 +34,7 @@ class ConstructOperator { const int64_t DEFAULT = 0; ConstructOperator() : dev_size_(0) {} ~ConstructOperator() = default; - Status Init(const RankList &dev_list, const Shape &dev_matrix_shape); + Status Init(const RankList &dev_list, const Shape &dev_matrix_shape, bool is_cost_model = false); OperatorVector SkipRedisReshapeOP(const Shape &shape); Status ReshapeOP(const Shape &shape); Status StridedSliceOP(const Args &args); @@ -51,6 +51,7 @@ class ConstructOperator { Shape tensor_shape_; RankList dev_list_; Shape dev_matrix_shape_; + bool is_cost_model_ = false; Status CreateGroupByDim(size_t axis, std::vector *group); }; } // namespace parallel diff --git a/mindspore/ccsrc/frontend/parallel/tensor_layout/redistribution_layout_transfer.cc b/mindspore/ccsrc/frontend/parallel/tensor_layout/redistribution_layout_transfer.cc index d8d09d95228..a7f8d1036e5 100644 --- a/mindspore/ccsrc/frontend/parallel/tensor_layout/redistribution_layout_transfer.cc +++ b/mindspore/ccsrc/frontend/parallel/tensor_layout/redistribution_layout_transfer.cc @@ -78,7 +78,9 @@ std::shared_ptr RedistributionLayoutTransfer::UnifyDevice std::shared_ptr RedistributionLayoutTransfer::UnifyDeviceArrangementAndTensorShape() const { std::shared_ptr unified_device_arrangement_ptr = UnifyDeviceArrangement(); if (unified_device_arrangement_ptr == nullptr) { - return nullptr; + ReshapeLayoutTransfer out; + out.SetExpandAble(false); + return std::make_shared(out); } Shape in_expand_shape; Status status = ExpandShape(unified_device_arrangement_ptr->from_in().tensor_shape().array(), diff --git a/mindspore/ccsrc/frontend/parallel/tensor_layout/redistribution_operator_infer.cc b/mindspore/ccsrc/frontend/parallel/tensor_layout/redistribution_operator_infer.cc index d5d78284dba..1d37b86d3d0 100644 --- a/mindspore/ccsrc/frontend/parallel/tensor_layout/redistribution_operator_infer.cc +++ b/mindspore/ccsrc/frontend/parallel/tensor_layout/redistribution_operator_infer.cc @@ -41,7 +41,7 @@ Status RedistributionOperatorInfer::Init(const TensorLayout &tensor_layout, cons operator_vector_.clear(); output_info_vector_.clear(); - if (constructor_.Init(dev_list_, dev_mat_.array()) != Status::SUCCESS) { + if (constructor_.Init(dev_list_, dev_mat_.array(), is_cost_model) != Status::SUCCESS) { MS_LOG(ERROR) << "Init constructor failed"; return Status::FAILED; } diff --git a/mindspore/ccsrc/frontend/parallel/tensor_layout/shape_util.cc b/mindspore/ccsrc/frontend/parallel/tensor_layout/shape_util.cc index 06edf604afc..734eee74245 100644 --- a/mindspore/ccsrc/frontend/parallel/tensor_layout/shape_util.cc +++ b/mindspore/ccsrc/frontend/parallel/tensor_layout/shape_util.cc @@ -77,7 +77,7 @@ Status AccumulateProductToShape(const Shape &shape_accum, Shape *shape) { return Status::FAILED; } if ((*iter) % value != 0) { - MS_LOG(ERROR) << "shape_accum is not a accumulate product in ascending order"; + MS_LOG(INFO) << "shape_accum is not a accumulate product in ascending order"; return Status::FAILED; } shape->push_back(static_cast((*iter) / value)); diff --git a/mindspore/python/mindspore/nn/optim/adasum.py b/mindspore/python/mindspore/nn/optim/adasum.py index d654d2ff67e..f85fd59b631 100644 --- a/mindspore/python/mindspore/nn/optim/adasum.py +++ b/mindspore/python/mindspore/nn/optim/adasum.py @@ -416,7 +416,6 @@ class AdaSumByGradWrapCell(Cell): self.sync_tensor = Parameter(Tensor(0, dtype=mstype.int32)) def construct(self, grads): - """adasum algorithm process.""" adasum_res = self.adasum(grads) sync_tensor = F.depend(self.sync_tensor, adasum_res) sync_flag = P.AllReduce()(sync_tensor) @@ -462,7 +461,6 @@ class AdaSumByDeltaWeightWrapCell(Cell): self.scale = Tensor(1.0, dtype=mstype.float32) def construct(self, grads): - """adasum algorithm process.""" grad_clone = self.hyper_map(F.partial(_clone_weight, self.scale), self.parameters) grads = F.depend(grads, grad_clone) opt_result = self.optimizer(grads) diff --git a/tests/ut/cpp/parallel/tensor_layout/construct_operator_test.cc b/tests/ut/cpp/parallel/tensor_layout/construct_operator_test.cc index dc7ca26fae8..2c33c63a30b 100644 --- a/tests/ut/cpp/parallel/tensor_layout/construct_operator_test.cc +++ b/tests/ut/cpp/parallel/tensor_layout/construct_operator_test.cc @@ -68,7 +68,7 @@ void TestConstructOperator::SetUp() { Shape tensor_shape = {512, 1024}; Shape dev_matrix_shape = {2, 4, 8, 16, 1}; RankList used_dev_list = g_device_manager->GetDeviceListByStageId(0); - constructor.Init(used_dev_list, dev_matrix_shape); + constructor.Init(used_dev_list, dev_matrix_shape, false); constructor.UpdateTensorShape(tensor_shape); } diff --git a/tests/ut/python/parallel/test_device_num_not_power_of_2.py b/tests/ut/python/parallel/test_device_num_not_power_of_2.py new file mode 100644 index 00000000000..21079ad188a --- /dev/null +++ b/tests/ut/python/parallel/test_device_num_not_power_of_2.py @@ -0,0 +1,181 @@ +# Copyright 2022 Huawei Technologies Co., Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import pytest +import numpy as np +import mindspore as ms +from mindspore import context, Tensor, Parameter +from mindspore.common.api import _cell_graph_executor +from mindspore.nn import Cell, TrainOneStepCell, Momentum +from mindspore.ops import operations as P + + +class Net(Cell): + def __init__(self, strategy1=None, strategy2=None, strategy3=None): + super().__init__() + self.mul = P.Mul().shard(strategy1) + self.matmul = P.MatMul().shard(strategy2) + self.gather = P.Gather().shard(strategy3) + self.reduce_sum = P.ReduceSum() + self.mul_weight = Parameter(Tensor(np.ones([64 * 3, 32 * 3]), dtype=ms.float32), "w1") + self.matmul_weight = Parameter(Tensor(np.ones([32 * 3, 32 * 3]), dtype=ms.float32), "w2") + self.embedding_table = Parameter(Tensor(np.ones([64 * 3, 32 * 3]), dtype=ms.float32), "embedding_table") + + def construct(self, x, b): + out = self.gather(self.embedding_table, x, 0) + out = self.matmul(out, self.matmul_weight) + out = self.mul(out, self.mul_weight) + out = out + b + return self.reduce_sum(out) + + +class Net1(Cell): + def __init__(self): + super().__init__() + self.mul = P.Mul() + self.matmul1 = P.MatMul(transpose_a=False, transpose_b=False) + self.matmul2 = P.MatMul(transpose_a=False, transpose_b=True) + self.matmul3 = P.MatMul(transpose_a=False, transpose_b=True) + self.matmul4 = P.MatMul(transpose_a=False, transpose_b=False) + self.reduce_sum = P.ReduceSum() + self.matmul_weight1 = Parameter(Tensor(np.ones([32 * 3, 32 * 3]), dtype=ms.float32), "mat_w1") + self.matmul_weight2 = Parameter(Tensor(np.ones([32 * 3, 32 * 3]), dtype=ms.float32), "mat_w2") + self.matmul_weight3 = Parameter(Tensor(np.ones([32 * 3, 32 * 3]), dtype=ms.float32), "mat_w3") + self.matmul_weight4 = Parameter(Tensor(np.ones([32 * 3, 32 * 3]), dtype=ms.float32), "mat_w4") + + def construct(self, x, b): + out = self.matmul1(x, self.matmul_weight1) + out = self.matmul2(out, self.matmul_weight2) + out = self.matmul3(out, self.matmul_weight3) + out = self.matmul4(out, self.matmul_weight4) + out = self.mul(out, b) + return self.reduce_sum(out) + +_x = Tensor(np.ones([3 * 64]), dtype=ms.int32) +_b = Tensor(np.ones([64 * 3, 32 * 3]), dtype=ms.float32) + +_x1 = Tensor(np.ones([3 * 32, 3 * 32]), dtype=ms.float32) +_b1 = Tensor(np.ones([32 * 3]), dtype=ms.float32) + +def compile_net(net, change_input=False): + optimizer = Momentum(net.trainable_params(), learning_rate=0.1, momentum=0.9) + train_net = TrainOneStepCell(net, optimizer) + train_net.set_auto_parallel() + train_net.set_train() + if change_input: + _cell_graph_executor.compile(train_net, _x1, _b1) + else: + _cell_graph_executor.compile(train_net, _x, _b) + context.reset_auto_parallel_context() + + +def test_auto_parallel_device_num_24(): + """ + Feature: device num 24 in auto parallel. + Description: verify device_num 24 in auto parallel by mul/matmul/gather + Expectation: compile done without error. + """ + context.set_context(device_target="Ascend") + context.set_auto_parallel_context(parallel_mode="semi_auto_parallel", device_num=24, global_rank=0) + mul_strategy1 = ((3 * 8, 1), (3 * 8, 1)) + matmul_strategy2 = ((24, 1), (1, 1)) + gather_strategy3 = ((1, 1), (24,)) + net = Net(mul_strategy1, matmul_strategy2, gather_strategy3) + compile_net(net) + +def test_auto_parallel_device_num_24_1(): + """ + Feature: device num 24 in auto parallel. + Description: verify device_num 24 in auto parallel by mul/matmul/gather + Expectation: compile done without error. + """ + context.set_context(device_target="Ascend") + context.set_auto_parallel_context(parallel_mode="semi_auto_parallel", device_num=24, global_rank=0) + mul_strategy1 = ((3, 8), (3, 8)) + matmul_strategy2 = ((3, 8), (8, 1)) + gather_strategy3 = ((24, 1), (1,)) + net = Net(mul_strategy1, matmul_strategy2, gather_strategy3) + compile_net(net) + +def test_auto_parallel_device_num_24_2(): + """ + Feature: device num 24 in auto parallel. + Description: verify device_num 24 in auto parallel by mul/matmul/gather + Expectation: compile done without error. + """ + context.set_context(device_target="Ascend") + context.set_auto_parallel_context(parallel_mode="semi_auto_parallel", device_num=24, global_rank=0) + mul_strategy1 = ((8, 3), (8, 3)) + matmul_strategy2 = ((8, 3), (3, 1)) + gather_strategy3 = ((3, 1), (1,)) + net = Net(mul_strategy1, matmul_strategy2, gather_strategy3) + with pytest.raises(RuntimeError): + compile_net(net) + +def test_auto_parallel_device_num_24_3(): + """ + Feature: device num 24 in auto parallel. + Description: verify device_num 24 in auto parallel by mul/matmul/gather + Expectation: compile done without error. + """ + context.set_context(device_target="Ascend") + context.set_auto_parallel_context(parallel_mode="semi_auto_parallel", device_num=24, global_rank=0) + mul_strategy1 = ((8, 3), (8, 3)) + matmul_strategy2 = ((8, 1), (1, 3)) + gather_strategy3 = ((24, 1), (1,)) + net = Net(mul_strategy1, matmul_strategy2, gather_strategy3) + with pytest.raises(RuntimeError): + compile_net(net) + +def test_auto_parallel_device_num_24_dyn_search(): + """ + Feature: device num 24 in auto parallel. + Description: verify device_num 24 in auto parallel by mul/matmul/gather + Expectation: compile done without error. + """ + context.set_context(device_target="Ascend") + context.set_auto_parallel_context(parallel_mode="auto_parallel", device_num=24, global_rank=0, + search_mode="dynamic_programming") + mul_strategy1 = None + matmul_strategy2 = None + gather_strategy3 = None + net = Net(mul_strategy1, matmul_strategy2, gather_strategy3) + compile_net(net) + +def test_auto_parallel_device_num_24_prop_search(): + """ + Feature: device num 24 in auto parallel. + Description: verify device_num 24 in auto parallel by mul/matmul/gather + Expectation: compile done without error. + """ + context.set_context(device_target="Ascend") + context.set_auto_parallel_context(parallel_mode="auto_parallel", device_num=24, global_rank=0, + search_mode="sharding_propagation") + mul_strategy1 = None + matmul_strategy2 = ((3, 8), (8, 1)) + gather_strategy3 = None + net = Net(mul_strategy1, matmul_strategy2, gather_strategy3) + compile_net(net) + +def test_auto_parallel_device_num_24_dyn_search_1(): + """ + Feature: device num 24 in auto parallel. + Description: verify device_num 24 in auto parallel, by 4 matmul + Expectation: compile done without error. + """ + context.set_context(device_target="Ascend") + context.set_auto_parallel_context(parallel_mode="auto_parallel", device_num=24, global_rank=0, + search_mode="dynamic_programming") + net = Net1() + compile_net(net, True) diff --git a/tests/ut/python/parallel/test_mix_precision_hybrid_parallel.py b/tests/ut/python/parallel/test_mix_precision_hybrid_parallel.py index 759c7f4be67..a120554ab9e 100644 --- a/tests/ut/python/parallel/test_mix_precision_hybrid_parallel.py +++ b/tests/ut/python/parallel/test_mix_precision_hybrid_parallel.py @@ -74,7 +74,7 @@ def test_two_matmul(): return out context.set_auto_parallel_context(device_num=8, global_rank=0) - strategy1 = ((2, 2), (2, 1)) + strategy1 = ((1, 1), (1, 8)) strategy2 = ((4, 2), (2, 1)) strategy3 = ((1, 8), (8, 1)) strategy4 = ((2, 4), (4, 1)) diff --git a/tests/ut/python/parallel/test_pipeline_opt_detection.py b/tests/ut/python/parallel/test_pipeline_opt_detection.py index 4c0f7c16eff..9c789c4ba9c 100644 --- a/tests/ut/python/parallel/test_pipeline_opt_detection.py +++ b/tests/ut/python/parallel/test_pipeline_opt_detection.py @@ -119,12 +119,12 @@ def test_pipeline_split_stage0(): Description:pipeline opt detection Expectation:success """ - context.set_auto_parallel_context(device_num=8, global_rank=0, pipeline_stages=2) + context.set_auto_parallel_context(device_num=32, global_rank=0, pipeline_stages=2) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") data = Tensor(np.ones([32, 64]), dtype=ms.float32) label = Tensor(np.ones([64, 64]), dtype=ms.float32) - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) net = PipelineCell(PipelineSplit(strategy1, strategy2), 4) params = net.trainable_params() dataset = DatasetLenet(data, label, 3) @@ -141,12 +141,12 @@ def test_pipeline_split_stage1(): Description:pipeline opt detection Expectation:success """ - context.set_auto_parallel_context(device_num=8, global_rank=4, pipeline_stages=2) + context.set_auto_parallel_context(device_num=32, global_rank=16, pipeline_stages=2) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") data = Tensor(np.ones([32, 64]), dtype=ms.float32) label = Tensor(np.ones([64, 64]), dtype=ms.float32) - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) net = PipelineCell(PipelineSplit(strategy1, strategy2), 4) params = net.trainable_params() dataset = DatasetLenet(data, label, 4) @@ -164,12 +164,12 @@ def test_pipeline_split_shared_parameter_stage0(): Description:pipeline opt detection Expectation:success """ - context.set_auto_parallel_context(device_num=8, global_rank=0, pipeline_stages=2) + context.set_auto_parallel_context(device_num=32, global_rank=0, pipeline_stages=2) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") data = Tensor(np.ones([32, 64]), dtype=ms.float32) label = Tensor(np.ones([64, 64]), dtype=ms.float32) - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) net = PipelineCell(PipelineSplitSharedParam(strategy1, strategy2), 4) params = net.trainable_params() dataset = DatasetLenet(data, label, 6) @@ -184,12 +184,12 @@ def test_pipeline_split_shared_parameter_stage1(): Description:pipeline opt detection Expectation:success """ - context.set_auto_parallel_context(device_num=8, global_rank=4, pipeline_stages=2) + context.set_auto_parallel_context(device_num=32, global_rank=16, pipeline_stages=2) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") data = Tensor(np.ones([32, 64]), dtype=ms.float32) label = Tensor(np.ones([64, 64]), dtype=ms.float32) - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) net = PipelineCell(PipelineSplitSharedParam(strategy1, strategy2), 4) params = net.trainable_params() dataset = DatasetLenet(data, label, 7) @@ -204,12 +204,12 @@ def test_pipeline_split_stage0_opt_shard(): Description:pipeline opt detection Expectation:success """ - context.set_auto_parallel_context(device_num=8, global_rank=0, pipeline_stages=2, enable_parallel_optimizer=True) + context.set_auto_parallel_context(device_num=32, global_rank=0, pipeline_stages=2, enable_parallel_optimizer=True) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") data = Tensor(np.ones([32, 64]), dtype=ms.float32) label = Tensor(np.ones([64, 64]), dtype=ms.float32) - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) net = PipelineCell(PipelineSplit(strategy1, strategy2), 4) params = net.trainable_params() dataset = DatasetLenet(data, label, 6) @@ -227,12 +227,12 @@ def test_pipeline_split_stage1_opt_shard(): Description:pipeline opt detection Expectation:success """ - context.set_auto_parallel_context(device_num=8, global_rank=4, pipeline_stages=2, enable_parallel_optimizer=True) + context.set_auto_parallel_context(device_num=32, global_rank=16, pipeline_stages=2, enable_parallel_optimizer=True) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") data = Tensor(np.ones([32, 64]), dtype=ms.float32) label = Tensor(np.ones([64, 64]), dtype=ms.float32) - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) net = PipelineCell(PipelineSplit(strategy1, strategy2), 4) params = net.trainable_params() dataset = DatasetLenet(data, label, 8) @@ -250,12 +250,12 @@ def test_pipeline_split_shared_parameter_stage0_opt_shard(): Description:pipeline opt detection Expectation:success """ - context.set_auto_parallel_context(device_num=8, global_rank=0, pipeline_stages=2, enable_parallel_optimizer=True) + context.set_auto_parallel_context(device_num=32, global_rank=0, pipeline_stages=2, enable_parallel_optimizer=True) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") data = Tensor(np.ones([32, 64]), dtype=ms.float32) label = Tensor(np.ones([64, 64]), dtype=ms.float32) - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) net = PipelineCell(PipelineSplitSharedParam(strategy1, strategy2), 4) params = net.trainable_params() dataset = DatasetLenet(data, label, 2) @@ -270,12 +270,12 @@ def test_pipeline_split_shared_parameter_stage1_opt_shard(): Description:pipeline opt detection Expectation:success """ - context.set_auto_parallel_context(device_num=8, global_rank=4, pipeline_stages=2, enable_parallel_optimizer=True) + context.set_auto_parallel_context(device_num=32, global_rank=16, pipeline_stages=2, enable_parallel_optimizer=True) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") data = Tensor(np.ones([32, 64]), dtype=ms.float32) label = Tensor(np.ones([64, 64]), dtype=ms.float32) - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) net = PipelineCell(PipelineSplitSharedParam(strategy1, strategy2), 4) params = net.trainable_params() dataset = DatasetLenet(data, label, 9) @@ -290,12 +290,12 @@ def test_pipeline_split_with_micro_batch_interleaved_stage0(): Description: net with MicroBatchInterleaved in semi auto parallel. Expectation: success. """ - context.set_auto_parallel_context(device_num=8, global_rank=0, pipeline_stages=2) + context.set_auto_parallel_context(device_num=32, global_rank=0, pipeline_stages=2) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") data = Tensor(np.ones([32, 64]), dtype=ms.float32) label = Tensor(np.ones([64, 64]), dtype=ms.float32) - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) micro_batch_interleaved = 2 net = PipelineCell(MicroBatchInterleaved(PipelineSplit(strategy1, strategy2), micro_batch_interleaved), 4) params = net.trainable_params() @@ -314,12 +314,12 @@ def test_pipeline_split_with_micro_batch_interleaved_stage1(): Description: net with MicroBatchInterleaved in semi auto parallel. Expectation: success. """ - context.set_auto_parallel_context(device_num=8, global_rank=4, pipeline_stages=2) + context.set_auto_parallel_context(device_num=32, global_rank=16, pipeline_stages=2) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") data = Tensor(np.ones([32, 64]), dtype=ms.float32) label = Tensor(np.ones([64, 64]), dtype=ms.float32) - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) micro_batch_interleaved = 2 net = PipelineCell(MicroBatchInterleaved(PipelineSplit(strategy1, strategy2), micro_batch_interleaved), 4) params = net.trainable_params() @@ -338,12 +338,12 @@ def test_pipeline_split_shared_parameter_with_micro_batch_interleaved_stage0_opt Description: net with MicroBatchInterleaved in semi auto parallel. Expectation: success. """ - context.set_auto_parallel_context(device_num=8, global_rank=0, pipeline_stages=2, enable_parallel_optimizer=True) + context.set_auto_parallel_context(device_num=32, global_rank=0, pipeline_stages=2, enable_parallel_optimizer=True) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") data = Tensor(np.ones([32, 64]), dtype=ms.float32) label = Tensor(np.ones([64, 64]), dtype=ms.float32) - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) micro_batch_interleaved = 2 net = PipelineCell(MicroBatchInterleaved(PipelineSplitSharedParam(strategy1, strategy2), micro_batch_interleaved), 4) @@ -360,12 +360,12 @@ def test_pipeline_split_shared_parameter_with_micro_batch_interleaved_stage1_opt Description: net with MicroBatchInterleaved in semi auto parallel. Expectation: success. """ - context.set_auto_parallel_context(device_num=8, global_rank=4, pipeline_stages=2, enable_parallel_optimizer=True) + context.set_auto_parallel_context(device_num=32, global_rank=16, pipeline_stages=2, enable_parallel_optimizer=True) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") data = Tensor(np.ones([32, 64]), dtype=ms.float32) label = Tensor(np.ones([64, 64]), dtype=ms.float32) - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) micro_batch_interleaved = 2 net = PipelineCell(MicroBatchInterleaved(PipelineSplitSharedParam(strategy1, strategy2), micro_batch_interleaved), 4) diff --git a/tests/ut/python/parallel/test_pipeline_shared_parameter.py b/tests/ut/python/parallel/test_pipeline_shared_parameter.py index a70b8dacde6..f26d40c0033 100644 --- a/tests/ut/python/parallel/test_pipeline_shared_parameter.py +++ b/tests/ut/python/parallel/test_pipeline_shared_parameter.py @@ -98,12 +98,12 @@ class Net(nn.Cell): def test_pipeline_split_stage0(): - context.set_auto_parallel_context(device_num=8, global_rank=0, pipeline_stages=2) + context.set_auto_parallel_context(device_num=32, global_rank=0, pipeline_stages=2) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") data = Tensor(np.ones([32, 64]), dtype=ms.float32) label = Tensor(np.ones([64, 64]), dtype=ms.float32) - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) net = PipelineCell(Net(strategy1, strategy2), 4) params = net.network.cell1.trainable_params() dataset = DatasetLenet(data, label, 3) @@ -113,12 +113,12 @@ def test_pipeline_split_stage0(): def test_pipeline_split_stage1(): - context.set_auto_parallel_context(device_num=8, global_rank=4, pipeline_stages=2) + context.set_auto_parallel_context(device_num=32, global_rank=16, pipeline_stages=2) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") data = Tensor(np.ones([32, 64]), dtype=ms.float32) label = Tensor(np.ones([64, 64]), dtype=ms.float32) - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) net = PipelineCell(Net(strategy1, strategy2), 4) params = net.network.cell2.trainable_params() dataset = DatasetLenet(data, label, 3) diff --git a/tests/ut/python/parallel/test_pipeline_split.py b/tests/ut/python/parallel/test_pipeline_split.py index c867d0400cf..8219ed4b227 100644 --- a/tests/ut/python/parallel/test_pipeline_split.py +++ b/tests/ut/python/parallel/test_pipeline_split.py @@ -117,12 +117,12 @@ class PipelineSplit2(nn.Cell): def test_pipeline_split_stage0(): - context.set_auto_parallel_context(device_num=8, global_rank=0, pipeline_stages=2) + context.set_auto_parallel_context(device_num=32, global_rank=0, pipeline_stages=2) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") data = Tensor(np.ones([32, 64]), dtype=ms.float32) label = Tensor(np.ones([64, 64]), dtype=ms.float32) - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) net = PipelineCell(PipelineSplit(strategy1, strategy2), 4) params = net.network.cell.block[0].trainable_params() dataset = DatasetLenet(data, label, 3) @@ -134,12 +134,12 @@ def test_pipeline_split_stage0(): assert param.name != "cell.block.1.param1" def test_pipeline_split_stage1(): - context.set_auto_parallel_context(device_num=8, global_rank=4, pipeline_stages=2) + context.set_auto_parallel_context(device_num=32, global_rank=16, pipeline_stages=2) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") data = Tensor(np.ones([32, 64]), dtype=ms.float32) label = Tensor(np.ones([64, 64]), dtype=ms.float32) - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) net = PipelineCell(PipelineSplit(strategy1, strategy2), 4) params = net.network.cell.block[1].trainable_params() dataset = DatasetLenet(data, label, 3) @@ -152,12 +152,12 @@ def test_pipeline_split_stage1(): def test_pipeline_split_shared_parameter_stage0(): - context.set_auto_parallel_context(device_num=8, global_rank=0, pipeline_stages=2) + context.set_auto_parallel_context(device_num=32, global_rank=0, pipeline_stages=2) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") data = Tensor(np.ones([32, 64]), dtype=ms.float32) label = Tensor(np.ones([64, 64]), dtype=ms.float32) - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) net = PipelineCell(PipelineSplit2(strategy1, strategy2), 4) params = net.network.cell.block[0].trainable_params() dataset = DatasetLenet(data, label, 3) @@ -167,12 +167,12 @@ def test_pipeline_split_shared_parameter_stage0(): def test_pipeline_split_shared_parameter_stage1(): - context.set_auto_parallel_context(device_num=8, global_rank=4, pipeline_stages=2) + context.set_auto_parallel_context(device_num=32, global_rank=16, pipeline_stages=2) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") data = Tensor(np.ones([32, 64]), dtype=ms.float32) label = Tensor(np.ones([64, 64]), dtype=ms.float32) - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) net = PipelineCell(PipelineSplit2(strategy1, strategy2), 4) params = net.network.cell.block[1].trainable_params() dataset = DatasetLenet(data, label, 3) @@ -182,36 +182,36 @@ def test_pipeline_split_shared_parameter_stage1(): def test_pipeline_split_shared_parameter_stage0_predict(): - context.set_auto_parallel_context(device_num=8, global_rank=0, pipeline_stages=2, full_batch=True) + context.set_auto_parallel_context(device_num=32, global_rank=0, pipeline_stages=2, full_batch=True) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") data = Tensor(np.ones([32, 64]), dtype=ms.float32) label = Tensor(np.ones([64, 64]), dtype=ms.float32) - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) net = PipelineSplit2(strategy1, strategy2) model = Model(net) model.predict(data, label) def test_pipeline_split_shared_parameter_stage1_predict(): - context.set_auto_parallel_context(device_num=8, global_rank=4, pipeline_stages=2, full_batch=True) + context.set_auto_parallel_context(device_num=32, global_rank=16, pipeline_stages=2, full_batch=True) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") data = Tensor(np.ones([32, 64]), dtype=ms.float32) label = Tensor(np.ones([64, 64]), dtype=ms.float32) - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) net = PipelineSplit2(strategy1, strategy2) model = Model(net) model.predict(data, label) def test_pipeline_split_stage0_opt_shard(): - context.set_auto_parallel_context(device_num=8, global_rank=0, pipeline_stages=2, enable_parallel_optimizer=True) + context.set_auto_parallel_context(device_num=32, global_rank=0, pipeline_stages=2, enable_parallel_optimizer=True) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") data = Tensor(np.ones([32, 64]), dtype=ms.float32) label = Tensor(np.ones([64, 64]), dtype=ms.float32) - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) net = PipelineCell(PipelineSplit(strategy1, strategy2), 4) params = net.network.cell.block[0].trainable_params() dataset = DatasetLenet(data, label, 3) @@ -224,12 +224,12 @@ def test_pipeline_split_stage0_opt_shard(): def test_pipeline_split_stage1_opt_shard(): - context.set_auto_parallel_context(device_num=8, global_rank=4, pipeline_stages=2, enable_parallel_optimizer=True) + context.set_auto_parallel_context(device_num=32, global_rank=16, pipeline_stages=2, enable_parallel_optimizer=True) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") data = Tensor(np.ones([32, 64]), dtype=ms.float32) label = Tensor(np.ones([64, 64]), dtype=ms.float32) - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) net = PipelineCell(PipelineSplit(strategy1, strategy2), 4) params = net.network.cell.block[1].trainable_params() dataset = DatasetLenet(data, label, 3) @@ -242,12 +242,12 @@ def test_pipeline_split_stage1_opt_shard(): def test_pipeline_split_shared_parameter_stage0_opt_shard(): - context.set_auto_parallel_context(device_num=8, global_rank=0, pipeline_stages=2, enable_parallel_optimizer=True) + context.set_auto_parallel_context(device_num=32, global_rank=0, pipeline_stages=2, enable_parallel_optimizer=True) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") data = Tensor(np.ones([32, 64]), dtype=ms.float32) label = Tensor(np.ones([64, 64]), dtype=ms.float32) - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) net = PipelineCell(PipelineSplit2(strategy1, strategy2), 4) params = net.network.cell.block[0].trainable_params() dataset = DatasetLenet(data, label, 3) @@ -257,12 +257,12 @@ def test_pipeline_split_shared_parameter_stage0_opt_shard(): def test_pipeline_split_shared_parameter_stage1_opt_shard(): - context.set_auto_parallel_context(device_num=8, global_rank=4, pipeline_stages=2, enable_parallel_optimizer=True) + context.set_auto_parallel_context(device_num=32, global_rank=16, pipeline_stages=2, enable_parallel_optimizer=True) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") data = Tensor(np.ones([32, 64]), dtype=ms.float32) label = Tensor(np.ones([64, 64]), dtype=ms.float32) - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) net = PipelineCell(PipelineSplit2(strategy1, strategy2), 4) params = net.network.cell.block[1].trainable_params() dataset = DatasetLenet(data, label, 3) @@ -277,12 +277,12 @@ def test_pipeline_split_with_micro_batch_interleaved_stage0(): Description: net with MicroBatchInterleaved in semi auto parallel. Expectation: success. """ - context.set_auto_parallel_context(device_num=8, global_rank=0, pipeline_stages=2) + context.set_auto_parallel_context(device_num=32, global_rank=0, pipeline_stages=2) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") data = Tensor(np.ones([32, 64]), dtype=ms.float32) label = Tensor(np.ones([64, 64]), dtype=ms.float32) - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) micro_batch_interleaved = 2 net = PipelineCell(MicroBatchInterleaved(PipelineSplit(strategy1, strategy2), micro_batch_interleaved), 4) params = net.network.network.cell.block[0].trainable_params() @@ -301,12 +301,12 @@ def test_pipeline_split_with_micro_batch_interleaved_stage1(): Description: net with MicroBatchInterleaved in semi auto parallel. Expectation: success. """ - context.set_auto_parallel_context(device_num=8, global_rank=4, pipeline_stages=2) + context.set_auto_parallel_context(device_num=32, global_rank=16, pipeline_stages=2) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") data = Tensor(np.ones([32, 64]), dtype=ms.float32) label = Tensor(np.ones([64, 64]), dtype=ms.float32) - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) micro_batch_interleaved = 2 net = PipelineCell(MicroBatchInterleaved(PipelineSplit(strategy1, strategy2), micro_batch_interleaved), 4) params = net.network.network.cell.block[1].trainable_params() @@ -325,12 +325,12 @@ def test_pipeline_split_shared_parameter_with_micro_batch_interleaved_stage0_opt Description: net with MicroBatchInterleaved in semi auto parallel. Expectation: success. """ - context.set_auto_parallel_context(device_num=8, global_rank=0, pipeline_stages=2, enable_parallel_optimizer=True) + context.set_auto_parallel_context(device_num=32, global_rank=0, pipeline_stages=2, enable_parallel_optimizer=True) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") data = Tensor(np.ones([32, 64]), dtype=ms.float32) label = Tensor(np.ones([64, 64]), dtype=ms.float32) - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) micro_batch_interleaved = 2 net = PipelineCell(MicroBatchInterleaved(PipelineSplit2(strategy1, strategy2), micro_batch_interleaved), 4) params = net.network.network.cell.block[0].trainable_params() @@ -346,12 +346,12 @@ def test_pipeline_split_shared_parameter_with_micro_batch_interleaved_stage1_opt Description: net with MicroBatchInterleaved in semi auto parallel. Expectation: success. """ - context.set_auto_parallel_context(device_num=8, global_rank=4, pipeline_stages=2, enable_parallel_optimizer=True) + context.set_auto_parallel_context(device_num=32, global_rank=16, pipeline_stages=2, enable_parallel_optimizer=True) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") data = Tensor(np.ones([32, 64]), dtype=ms.float32) label = Tensor(np.ones([64, 64]), dtype=ms.float32) - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) micro_batch_interleaved = 2 net = PipelineCell(MicroBatchInterleaved(PipelineSplit2(strategy1, strategy2), micro_batch_interleaved), 4) params = net.network.network.cell.block[1].trainable_params() @@ -409,11 +409,11 @@ class TestPipelineSplitWithNoOptimizer: Expectation: the number of the float16 tensor is not equal to 16, 16 is obtained by manually checked graph. the number of the Mirror is not equal to 2, 2 is obtained by manually checked graph. """ - context.set_auto_parallel_context(device_num=8, global_rank=0, pipeline_stages=2, + context.set_auto_parallel_context(device_num=32, global_rank=0, pipeline_stages=2, enable_parallel_optimizer=False) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) pipeline_net = PipelineSplit(strategy1, strategy2, dtype=ms.float16) run_pipeline_split_function(pipeline_net, micro_batch_interleaved=1) self.cat_fp16_from_ir(pattern='grad_mirror_MirrorMicroStepOperator', @@ -428,14 +428,60 @@ class TestPipelineSplitWithNoOptimizer: Expectation: the number of the float16 tensor is not equal to 16, 16 is obtained by manually checked graph. the number of the Mirror is not equal to 2, 2 is obtained by manually checked graph. """ - context.set_auto_parallel_context(device_num=8, global_rank=0, pipeline_stages=2, + context.set_auto_parallel_context(device_num=32, global_rank=0, pipeline_stages=2, enable_parallel_optimizer=False) context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") - strategy1 = ((4, 1), (1, 1)) - strategy2 = ((2, 1), (1, 1)) + strategy1 = ((16, 1), (1, 1)) + strategy2 = ((8, 1), (1, 1)) pipeline_net = PipelineSplit(strategy1, strategy2, dtype=ms.float16) run_pipeline_split_function(pipeline_net, micro_batch_interleaved=2) self.cat_fp16_from_ir(pattern='grad_mirror_MirrorMicroStepOperator', target_count=2) self.cat_fp16_from_ir(pattern='Cast(', target_count=26) + +def test_pipeline_split_stage0_device_num_48(): + """ + Feature: test PipelineSplit with 48 devices in auto parallel. + Description: net with pipeline parallel in auto parallel mode using 48 devices, stage0. + Expectation: success. + """ + context.set_auto_parallel_context(device_num=48, global_rank=0, pipeline_stages=2) + context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") + context.set_context(device_target="Ascend") + data = Tensor(np.ones([32 * 6, 64]), dtype=ms.float32) + label = Tensor(np.ones([64 * 6, 64]), dtype=ms.float32) + strategy1 = ((3, 8), (8, 1)) + strategy2 = ((24, 1), (1, 1)) + net = PipelineCell(PipelineSplit(strategy1, strategy2), 4) + params = net.network.cell.block[0].trainable_params() + dataset = DatasetLenet(data, label, 3) + optimizer = nn.Lamb(params, learning_rate=0.01) + model = Model(net, optimizer=optimizer) + model.train(2, dataset, dataset_sink_mode=False) + for _, param in model._train_network.parameters_and_names(): + assert param.name != "cell.block.1.param" + assert param.name != "cell.block.1.param1" + +def test_pipeline_split_stage1_device_num_48(): + """ + Feature: test PipelineSplit with 48 devices in auto parallel. + Description: net with pipeline parallel in auto parallel mode using 48 devices, stage1. + Expectation: success. + """ + context.set_auto_parallel_context(device_num=48, global_rank=24, pipeline_stages=2) + context.set_auto_parallel_context(parallel_mode="semi_auto_parallel") + context.set_context(device_target="Ascend") + data = Tensor(np.ones([32 * 6, 64]), dtype=ms.float32) + label = Tensor(np.ones([64 * 6, 64]), dtype=ms.float32) + strategy1 = ((3, 8), (8, 1)) + strategy2 = ((24, 1), (1, 1)) + net = PipelineCell(PipelineSplit(strategy1, strategy2), 4) + params = net.network.cell.block[1].trainable_params() + dataset = DatasetLenet(data, label, 3) + optimizer = nn.Lamb(params, learning_rate=0.01) + model = Model(net, optimizer=optimizer) + model.train(2, dataset, dataset_sink_mode=False) + for _, param in model._train_network.parameters_and_names(): + assert param.name != "cell.block.0.param" + assert param.name != "cell.block.0.param1" diff --git a/tests/ut/python/parallel/test_tile.py b/tests/ut/python/parallel/test_tile.py index f25219bcb0a..6b56c33f58e 100644 --- a/tests/ut/python/parallel/test_tile.py +++ b/tests/ut/python/parallel/test_tile.py @@ -119,9 +119,9 @@ def test_tile_tensor_no_full_split(): def test_tile_tensor_no_full_split2(): - context.set_auto_parallel_context(parallel_mode="semi_auto_parallel", device_num=8, global_rank=0) - strategy1 = ((2, 2, 1), (2, 2, 1)) - strategy2 = ((2, 2, 1),) + context.set_auto_parallel_context(parallel_mode="semi_auto_parallel", device_num=32, global_rank=0) + strategy1 = ((4, 4, 1), (4, 4, 1)) + strategy2 = ((4, 4, 1),) net = Net3(_w1, strategy1, strategy2) compile_net(net, _x1, _b)