yankf21
/
mindspore2022
forked from huawei/mindspore2022
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

support not only power of 2

This commit is contained in:
yao_yf 2022-03-04 15:05:28 +08:00
parent bd0d719fdf
commit b60e54e0d5
39 changed files with 756 additions and 225 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

89
mindspore/ccsrc/frontend/parallel/device_manager.cc
View File

@ -19,6 +19,7 @@
#include <algorithm>
#include <string>
#include <vector>
#include <unordered_map>
#include "utils/hash_set.h"
#include "frontend/parallel/step_parallel.h"
@ -160,6 +161,67 @@ std::shared_ptr<Device> GetListMemberByIndex(size_t index, const std::vector<std
return result;
}
namespace {
constexpr size_t NODE_PER_SERVER = 8;
Status IsFeasibleDeiveListOneServer(const RankList &rank_list) {
if (rank_list.size() == 1 || rank_list.size() == 8) {
return SUCCESS;
}
if (rank_list.size() == 4 && (rank_list[3] - rank_list[0] == 3) && (rank_list[0] == 0 || rank_list[3] == 7)) {
return SUCCESS;
}
if (rank_list.size() == 4 && (rank_list[3] % 2 == rank_list[2] % 2) && (rank_list[2] % 2 == rank_list[1] % 2) &&
(rank_list[1] % 2 == rank_list[0] % 2)) {
return SUCCESS;
}
if (rank_list.size() == 2) {
if (rank_list[1] - rank_list[0] == 4) {
return SUCCESS;
}
if (rank_list[1] < 4 && rank_list[0] < 4) {
return SUCCESS;
}
if (rank_list[1] >= 4 && rank_list[0] >= 4) {
return SUCCESS;
}
}
return FAILED;
}
Status IsFeasibleDeiveList(const RankList &rank_list) {
std::unordered_map<int64_t, RankList> 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<RankList> 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<std::string>(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<mindspore::parallel::Device> &devices) {
Group g;
(void)gm_.CreateGroup(group_name, devices, &g);
return g;
Status DeviceManager::CreateGroup(const std::string &group_name,
const std::vector<mindspore::parallel::Device> &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<int64_t> 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() {

5
mindspore/ccsrc/frontend/parallel/device_manager.h
View File

@ -71,8 +71,8 @@ class DeviceManager {
Device CreateNewDeviceByRank(int64_t rank) const;
std::vector<Device> CreateDeviceListByRankList(RankList ranks);
std::string GenerateGroupNameByRanks(RankList dev_ranks);
Group CreateGroup(const std::string &group_name, const std::vector<Device> &devices);
Group CreateGroup(const RankList &dev_ranks);
Status CreateGroup(const std::string &group_name, const std::vector<Device> &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<std::pair<std::string, std::vector<uint32_t>>> 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<std::shared_ptr<Device>> devices_;

8
mindspore/ccsrc/frontend/parallel/ops_info/activation_info.cc
View File

@ -280,7 +280,7 @@ Status CumSumInfo::InferMirrorOps() {
Shape input_a_tensor_map = inputs_tensor_map_.at(0);
std::vector<Group> 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> 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> 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> group;
if (CreateGroupByTensorMap(inputs_tensor_map_[0], &group) != SUCCESS) {
MS_LOG(ERROR) << name_ << ": Create group failed";
ReportError(name_ + ": Create group failed.");
return FAILED;
}

8
mindspore/ccsrc/frontend/parallel/ops_info/batchnorm_info.cc
View File

@ -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> group_list;
if (CreateGroupByTensorMap(tmp_map, &group_list) != SUCCESS) {
MS_LOG(ERROR) << name_ << ": Create group failed";
ReportError(name_ + ": Create group failed.");
return FAILED;
}

4
mindspore/ccsrc/frontend/parallel/ops_info/conv2d_info.cc
View File

@ -651,7 +651,7 @@ Status Conv2DInfo::InferForwardCommunication() {
std::vector<Group> 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> 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;
}

5
mindspore/ccsrc/frontend/parallel/ops_info/crop_and_resize_info.cc
View File

@ -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;
}

9
mindspore/ccsrc/frontend/parallel/ops_info/gather_info.cc
View File

@ -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<Group> 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;
}

2
mindspore/ccsrc/frontend/parallel/ops_info/gatherd_info.cc
View File

@ -154,7 +154,7 @@ Status GatherDInfo::InferMirrorOps() {
std::vector<Group> group;
if (CreateGroupByTensorMap(inputs_tensor_map_[0], &group) != SUCCESS) {
MS_LOG(ERROR) << name_ << ": Create group failed";
ReportError(name_ + ": Create group failed.");
return FAILED;
}

152
mindspore/ccsrc/frontend/parallel/ops_info/matmul_info.cc
View File

@ -262,7 +262,7 @@ Status MatMulBase::InferForwardCommunication() {
std::vector<Group> 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<StrategyPtr> *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<void(uint64_t, size_t)> 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<StrategyPtr> &sp_vector_2_power_part) {
std::vector<StrategyPtr> 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<Strategy>(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<void(uint64_t, size_t)> 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<StrategyPtr> 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<StrategyPtr> MatMulBase::GenerateOpStrategies(int64_t) {
@ -601,7 +661,7 @@ std::shared_ptr<Strategys> 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);

4
mindspore/ccsrc/frontend/parallel/ops_info/matmul_info.h
View File

@ -53,6 +53,10 @@ class MatMulBase : public OperatorInfo {
Status InferTensorMap() override;
Status InferTensorLayout(TensorLayouts *inputs_layout, TensorLayouts *outputs_layout);
void InitTensorInfoForCost(std::vector<TensorInfo> *);
Status GenerateStrategiesBase(int64_t stage_id, size_t dev_num, const Shape &input0_shape, Shape input1_shape,
std::vector<StrategyPtr> *const sp_vector);
Status GenerateStrategiesNotPower2(int64_t stage_id, size_t dev_num_not_2_power,
const std::vector<StrategyPtr> &sp_vector_2_power_part);
Status CheckForTensorSliceValid() const;
Status GetAttrs() override;

157
mindspore/ccsrc/frontend/parallel/ops_info/operator_info.cc
View File

@ -187,7 +187,7 @@ Status OperatorInfo::InferMirrorOps() {
for (size_t i = 0; i < inputs_tensor_map_.size(); ++i) {
std::vector<Group> 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> *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<StrategyPtr> *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<StrategyPtr> *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<StrategyPtr> *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<StrategyPtr> 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<Strategy>(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])

16
mindspore/ccsrc/frontend/parallel/ops_info/operator_info.h
View File

@ -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<ValuePtr> &input_value) { input_value_ = input_value; }
const std::vector<ValuePtr> &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> *group);
Status CreateGroupForOptShard(TensorLayout *const tensor_layout, std::vector<Group> *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<int64_t> 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<Strategys> GenerateBatchStrategiesBySplitFlag(const Shapes &shap
const std::vector<bool> &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<StrategyPtr> *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<StrategyPtr> *sp_vector);

8
mindspore/ccsrc/frontend/parallel/ops_info/reduce_method_info.cc
View File

@ -186,7 +186,7 @@ Status ReduceMethod::InferForwardCommunication() {
}
std::vector<Group> 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<Group> 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<Group> 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<Group> input_group;
if (CreateGroupByTensorMap(input_tensor_map, &input_group) != SUCCESS) {
MS_LOG(ERROR) << name_ << ": Infer MirrorOps failed.";
ReportError(name_ + ": Create group failed.");
return FAILED;
}

2
mindspore/ccsrc/frontend/parallel/ops_info/reshape_info.cc
View File

@ -51,7 +51,7 @@ Status ReshapeInfo::InferMirrorOps() {
Shape input_tensor_map = input_layout_.tensor_map().array();
std::vector<Group> input_group;
if (CreateGroupByTensorMap(input_tensor_map, &input_group) != SUCCESS) {
MS_LOG(ERROR) << name_ << ": Infer MirrorOps failed.";
ReportError(name_ + ": Create group failed.");
return FAILED;
}

4
mindspore/ccsrc/frontend/parallel/ops_info/roi_align_info.cc
View File

@ -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;
}

2
mindspore/ccsrc/frontend/parallel/ops_info/slice_info.cc
View File

@ -131,7 +131,7 @@ Status SliceInfo::InferMirrorOps() {
Shape input_tensor_map = inputs_tensor_map_[0];
std::vector<Group> group;
if (CreateGroupByTensorMap(input_tensor_map, &group) != SUCCESS) {
MS_LOG(ERROR) << name_ << ": Create group for input failed.";
ReportError(name_ + ": Create group failed.");
return FAILED;
}

2
mindspore/ccsrc/frontend/parallel/ops_info/strided_slice_info.cc
View File

@ -166,7 +166,7 @@ Status StridedSliceInfo::InferMirrorOps() {
Shape input_tensor_map = inputs_tensor_map_[0];
std::vector<Group> group;
if (CreateGroupByTensorMap(input_tensor_map, &group) != SUCCESS) {
MS_LOG(ERROR) << name_ << ": Create group for input failed.";
ReportError(name_ + ": Create group failed.");
return FAILED;
}

2
mindspore/ccsrc/frontend/parallel/ops_info/tensordot_info.cc
View File

@ -196,7 +196,7 @@ Status TensorDotInfo::InferForwardCommunication() {
std::vector<Group> 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;
}

2
mindspore/ccsrc/frontend/parallel/ops_info/tile_info.cc
View File

@ -139,7 +139,7 @@ Status TileInfo::InferMirrorOps() {
Shape input_tensor_map = inputs_tensor_map_[0];
std::vector<Group> group;
if (CreateGroupByTensorMap(input_tensor_map, &group) != SUCCESS) {
MS_LOG(ERROR) << name_ << ": Create group for input failed.";
ReportError(name_ + ": Create group failed.");
return FAILED;
}

2
mindspore/ccsrc/frontend/parallel/ops_info/topk_info.cc
View File

@ -120,7 +120,7 @@ Status TopKInfo::InferMirrorOps() {
for (size_t i = 0; i < inputs_tensor_map_.size(); ++i) {
std::vector<Group> 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;
}

4
mindspore/ccsrc/frontend/parallel/ops_info/unsorted_segment_op_info.cc
View File

@ -104,7 +104,7 @@ Status UnsortedSegmentOpInfo::InferMirrorOps() {
Shape tensor_map = inputs_tensor_map_[0];
std::vector<Group> 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.";

15
mindspore/ccsrc/frontend/parallel/parameter_manager.cc
View File

@ -764,7 +764,10 @@ std::vector<bool> 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<int64_t>(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<int64_t>(new_group.size()));

12
mindspore/ccsrc/frontend/parallel/pipeline_transformer/pipeline_transformer.cc
View File

@ -181,9 +181,17 @@ void PipelineTransformer::LabelMicroBatch() {
void PipelineTransformer::CreateForwardGroup() {
std::vector<int64_t> 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());
}

6
mindspore/ccsrc/frontend/parallel/step_auto_parallel.cc
View File

@ -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

14
mindspore/ccsrc/frontend/parallel/step_parallel.cc
View File

@ -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<CNodePtr>(), 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<CNodePtr>(), group_between_stages.name(), PARALLEL_GLOBALNORM_BETWEEN);
}
}

4
mindspore/ccsrc/frontend/parallel/tensor_layout/arrangement.cc
View File

@ -115,7 +115,7 @@ std::shared_ptr<std::vector<Arrangement>> 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<std::vector<Arrangement>> 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<std::vector<Arrangement>>(arrangement_list);

13
mindspore/ccsrc/frontend/parallel/tensor_layout/construct_operator.cc
View File

@ -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<Group> *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;
}

3
mindspore/ccsrc/frontend/parallel/tensor_layout/construct_operator.h
View File

@ -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> *group);
};
} // namespace parallel

4
mindspore/ccsrc/frontend/parallel/tensor_layout/redistribution_layout_transfer.cc
View File

@ -78,7 +78,9 @@ std::shared_ptr<ReshapeLayoutTransfer> RedistributionLayoutTransfer::UnifyDevice
std::shared_ptr<ReshapeLayoutTransfer> RedistributionLayoutTransfer::UnifyDeviceArrangementAndTensorShape() const {
std::shared_ptr<ReshapeLayoutTransfer> unified_device_arrangement_ptr = UnifyDeviceArrangement();
if (unified_device_arrangement_ptr == nullptr) {
return nullptr;
ReshapeLayoutTransfer out;
out.SetExpandAble(false);
return std::make_shared<ReshapeLayoutTransfer>(out);
}
Shape in_expand_shape;
Status status = ExpandShape(unified_device_arrangement_ptr->from_in().tensor_shape().array(),

2
mindspore/ccsrc/frontend/parallel/tensor_layout/redistribution_operator_infer.cc
View File

@ -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;
}

2
mindspore/ccsrc/frontend/parallel/tensor_layout/shape_util.cc
View File

@ -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<int64_t>((*iter) / value));

2
mindspore/python/mindspore/nn/optim/adasum.py
View File

@ -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)

2
tests/ut/cpp/parallel/tensor_layout/construct_operator_test.cc
View File

@ -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);
}

181
tests/ut/python/parallel/test_device_num_not_power_of_2.py Normal file
View File

@ -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)

2
tests/ut/python/parallel/test_mix_precision_hybrid_parallel.py
View File

@ -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))

72
tests/ut/python/parallel/test_pipeline_opt_detection.py
View File

@ -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)

12
tests/ut/python/parallel/test_pipeline_shared_parameter.py
View File

@ -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)

142
tests/ut/python/parallel/test_pipeline_split.py
View File

@ -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"

6
tests/ut/python/parallel/test_tile.py
View File

@ -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)