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

[bugfix] use group rank id replace global rank id for broadcast unique id

This commit is contained in:
lizhenyu 2022-03-29 19:24:43 +08:00
parent 79f4a2b1a7
commit de60949038
6 changed files with 17 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
mindspore/ccsrc/distributed/collective/collective_manager.cc
View File

@ -195,7 +195,6 @@ bool CollectiveManager::CreateCommunicationGroup(const std::string &group_name,
// Step 3: Generate device information of the root node.
CommunicationGroupPtr group = device_comm_lib_instance_->GetGroup(group_name);
MS_EXCEPTION_IF_NULL(group);
bool is_root_node = (group->GetGroupRank(global_rank_id_) == 0);
size_t root_info_size = 0;
void *root_info = group->GenerateRootInfo(&root_info_size);
MS_EXCEPTION_IF_NULL(root_info);
@ -203,7 +202,7 @@ bool CollectiveManager::CreateCommunicationGroup(const std::string &group_name,
bool ret = false;
// Step 4: Broadcast the device root information to all nodes on host side.
while (!ret) {
ret = host_comm_lib_instance_->BroadcastUniqueID(group_name, is_root_node, root_info_size, root_info);
ret = host_comm_lib_instance_->BroadcastUniqueID(group_name, root_info_size, root_info);
if (!ret) {
MS_LOG(ERROR) << "Broadcast for device root info failed on the host side.";
return false;

13
mindspore/ccsrc/plugin/device/cpu/hal/hardware/ms_collective_comm_lib.cc
View File

@ -73,10 +73,12 @@ bool MsCollectiveCommLib::AllGatherHostHashName(size_t host_hash_name, std::vect
return true;
}
bool MsCollectiveCommLib::BroadcastUniqueID(const std::string &group_name, bool is_root_node, size_t root_info_size,
void *root_info) const {
bool MsCollectiveCommLib::BroadcastUniqueID(const std::string &group_name, size_t root_info_size, void *root_info) {
CHECK_IF_NULL(root_info);
if (is_root_node) {
auto group = GetGroup(group_name);
CHECK_IF_NULL(group);
uint32_t group_rank_id = group->GetGroupRank(node_->rank_id());
if (group_rank_id == 0) {
while (!SendUniqueID(group_name, root_info_size, root_info)) {
MS_LOG(WARNING) << "Send unique id to scheduler failed, retrying...";
if (finalized_.load()) {
@ -160,11 +162,13 @@ bool MsCollectiveCommLib::SendUniqueID(const std::string &group_name, size_t roo
const void *root_info) const {
CHECK_IF_NULL(root_info);
CHECK_IF_NULL(node_);
ps::core::SendUniqueIDMessage send_unique_id_msg;
send_unique_id_msg.set_node_id(node_->node_id());
send_unique_id_msg.set_rank_id(node_->rank_id());
send_unique_id_msg.set_rank_id(0);
send_unique_id_msg.set_group_name(group_name);
send_unique_id_msg.set_unique_id(root_info, root_info_size);
std::shared_ptr<std::vector<unsigned char>> output = nullptr;
if (!node_->SendToScheduler(send_unique_id_msg.SerializeAsString().data(),
send_unique_id_msg.SerializeAsString().size(), NodeCommand::SEND_UNIQUE_ID, &output)) {
@ -187,7 +191,6 @@ bool MsCollectiveCommLib::QueryUniqueID(const std::string &group_name, size_t ro
CHECK_IF_NULL(node_);
ps::core::QueryUniqueIDMessage query_unique_id_msg;
query_unique_id_msg.set_node_id(node_->node_id());
query_unique_id_msg.set_rank_id(node_->rank_id());
query_unique_id_msg.set_group_name(group_name);
std::shared_ptr<std::vector<unsigned char>> output = nullptr;
if (!node_->SendToScheduler(query_unique_id_msg.SerializeAsString().data(),

3
mindspore/ccsrc/plugin/device/cpu/hal/hardware/ms_collective_comm_lib.h
View File

@ -51,8 +51,7 @@ class MsCollectiveCommLib : public CollectiveCommunicationLib {
bool AllGatherHostHashName(size_t host_hash_name, std::vector<size_t> *host_hash_names) const override;
bool BroadcastUniqueID(const std::string &group_name, bool is_root_node, size_t root_info_size,
void *root_info) const override;
bool BroadcastUniqueID(const std::string &group_name, size_t root_info_size, void *root_info) override;
bool AllGather(const void *send_buff, void *recv_buff, size_t send_count, TypeId data_type,
const std::string &group_name, void *stream = nullptr) override;

4
mindspore/ccsrc/ps/core/protos/comm.proto
View File

@ -292,10 +292,8 @@ message SendUniqueIDMessage {
message QueryUniqueIDMessage {
// The unique node id.
string node_id = 1;
// The rank id of the node in the cluster.
uint32 rank_id = 2;
// The group name of goupt which need to initialize collective communication.
string group_name = 3;
string group_name = 2;
}
message QueryUniqueIDRespMessage {

13
mindspore/ccsrc/ps/core/ps_scheduler_node.cc
View File

@ -152,12 +152,12 @@ void PSSchedulerNode::ProcessSendUniqueID(const std::shared_ptr<TcpServer> &serv
uint32_t rank_id = send_unique_id_msg.rank_id();
std::string group_name = send_unique_id_msg.group_name();
MS_LOG(INFO) << "Receive send unique id request, group name: " << group_name << ", node id: " << node_id
<< ", rank id: " << rank_id;
<< ", group rank id: " << rank_id;
bool ret = false;
std::string error = "";
if (rank_id != 0) {
error = "The rank id: " + std::to_string(rank_id) + " of worker which sends unique id should be 0";
error = "The group rank id: " + std::to_string(rank_id) + " of worker which sends unique id should be 0";
MS_LOG(ERROR) << error;
} else {
unique_id_group_[group_name] = send_unique_id_msg.unique_id();
@ -166,7 +166,7 @@ void PSSchedulerNode::ProcessSendUniqueID(const std::shared_ptr<TcpServer> &serv
GeneralResponse(server, conn, meta, ret, error);
MS_LOG(INFO) << "Respond send unique id request, group name: " << group_name << ", node id: " << node_id
<< ", rank id: " << rank_id;
<< ", group rank id: " << rank_id;
}
void PSSchedulerNode::ProcessQueryUniqueID(const std::shared_ptr<TcpServer> &server,
@ -180,10 +180,8 @@ void PSSchedulerNode::ProcessQueryUniqueID(const std::shared_ptr<TcpServer> &ser
QueryUniqueIDMessage query_msg;
query_msg.ParseFromArray(data, SizeToInt(size));
std::string node_id = query_msg.node_id();
uint32_t rank_id = query_msg.rank_id();
std::string group_name = query_msg.group_name();
MS_LOG(INFO) << "Receive query unique id request, group name: " << group_name << ", node id: " << node_id
<< ", rank id: " << rank_id;
MS_LOG(INFO) << "Receive query unique id request, group name: " << group_name << ", node id: " << node_id;
auto iter = unique_id_group_.find(group_name);
bool is_success = (iter != unique_id_group_.end());
@ -200,8 +198,7 @@ void PSSchedulerNode::ProcessQueryUniqueID(const std::shared_ptr<TcpServer> &ser
return;
}
MS_LOG(INFO) << "Respond query unique id request, group name: " << group_name << ", node id: " << node_id
<< ", rank id: " << rank_id;
MS_LOG(INFO) << "Respond query unique id request, group name: " << group_name << ", node id: " << node_id;
}
void PSSchedulerNode::ProcessSendFinishTransform(const std::shared_ptr<TcpServer> &server,

5
mindspore/ccsrc/runtime/collective/collective_communication_lib.h
View File

@ -83,10 +83,7 @@ class CollectiveCommunicationLib {
virtual bool AllGatherHostHashName(size_t host_hash_name, std::vector<size_t> *host_hash_names) const { return true; }
// Broadcast the device root information to all nodes on host side, used to initialize collective communication.
virtual bool BroadcastUniqueID(const std::string &group_name, bool is_root_node, size_t root_info_size,
void *root_info) const {
return true;
}
virtual bool BroadcastUniqueID(const std::string &group_name, size_t root_info_size, void *root_info) { return true; }
// Primitive of collective operations.
virtual bool AllGather(const void *send_buff, void *recv_buff, size_t send_count, TypeId data_type,