diff --git a/mindspore/ccsrc/distributed/collective/collective_manager.cc b/mindspore/ccsrc/distributed/collective/collective_manager.cc index efa07c7e259..3f4b882a4a4 100644 --- a/mindspore/ccsrc/distributed/collective/collective_manager.cc +++ b/mindspore/ccsrc/distributed/collective/collective_manager.cc @@ -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; diff --git a/mindspore/ccsrc/plugin/device/cpu/hal/hardware/ms_collective_comm_lib.cc b/mindspore/ccsrc/plugin/device/cpu/hal/hardware/ms_collective_comm_lib.cc index 053d957f720..09f414ca539 100644 --- a/mindspore/ccsrc/plugin/device/cpu/hal/hardware/ms_collective_comm_lib.cc +++ b/mindspore/ccsrc/plugin/device/cpu/hal/hardware/ms_collective_comm_lib.cc @@ -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> 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> output = nullptr; if (!node_->SendToScheduler(query_unique_id_msg.SerializeAsString().data(), diff --git a/mindspore/ccsrc/plugin/device/cpu/hal/hardware/ms_collective_comm_lib.h b/mindspore/ccsrc/plugin/device/cpu/hal/hardware/ms_collective_comm_lib.h index 01c9489bdba..8e67b75ec31 100644 --- a/mindspore/ccsrc/plugin/device/cpu/hal/hardware/ms_collective_comm_lib.h +++ b/mindspore/ccsrc/plugin/device/cpu/hal/hardware/ms_collective_comm_lib.h @@ -51,8 +51,7 @@ class MsCollectiveCommLib : public CollectiveCommunicationLib { bool AllGatherHostHashName(size_t host_hash_name, std::vector *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; diff --git a/mindspore/ccsrc/ps/core/protos/comm.proto b/mindspore/ccsrc/ps/core/protos/comm.proto index ad788f459c6..cb24b945aaf 100644 --- a/mindspore/ccsrc/ps/core/protos/comm.proto +++ b/mindspore/ccsrc/ps/core/protos/comm.proto @@ -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 { diff --git a/mindspore/ccsrc/ps/core/ps_scheduler_node.cc b/mindspore/ccsrc/ps/core/ps_scheduler_node.cc index 5e368f693aa..5d4f492b9d1 100644 --- a/mindspore/ccsrc/ps/core/ps_scheduler_node.cc +++ b/mindspore/ccsrc/ps/core/ps_scheduler_node.cc @@ -152,12 +152,12 @@ void PSSchedulerNode::ProcessSendUniqueID(const std::shared_ptr &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 &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 &server, @@ -180,10 +180,8 @@ void PSSchedulerNode::ProcessQueryUniqueID(const std::shared_ptr &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 &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 &server, diff --git a/mindspore/ccsrc/runtime/collective/collective_communication_lib.h b/mindspore/ccsrc/runtime/collective/collective_communication_lib.h index f272e28e5f1..b271d5910f7 100644 --- a/mindspore/ccsrc/runtime/collective/collective_communication_lib.h +++ b/mindspore/ccsrc/runtime/collective/collective_communication_lib.h @@ -83,10 +83,7 @@ class CollectiveCommunicationLib { virtual bool AllGatherHostHashName(size_t host_hash_name, std::vector *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,