From 35e4e9aed687493ad36272dc7c102a7e01d5cbfd Mon Sep 17 00:00:00 2001 From: ZPaC Date: Tue, 29 Mar 2022 15:17:13 +0800 Subject: [PATCH] Encapsulate methods and optimize logs. --- .../parallel/graph_util/graph_splitter.cc | 211 ++++++++++-------- .../parallel/graph_util/graph_splitter.h | 13 ++ .../device/cpu/kernel/rpc/rpc_send_kernel.cc | 3 +- .../device/gpu/hal/device/gpu_comm_manager.cc | 4 +- .../graph_scheduler/actor/rpc/recv_actor.cc | 13 ++ .../graph_scheduler/actor/rpc/recv_actor.h | 12 +- .../graph_scheduler/actor/rpc/rpc_actor.cc | 8 +- .../graph_scheduler/actor/rpc/rpc_actor.h | 4 +- .../graph_scheduler/actor/rpc/send_actor.cc | 3 +- .../graph_scheduler/rpc_node_scheduler.cc | 8 +- 10 files changed, 173 insertions(+), 106 deletions(-) diff --git a/mindspore/ccsrc/frontend/parallel/graph_util/graph_splitter.cc b/mindspore/ccsrc/frontend/parallel/graph_util/graph_splitter.cc index eafd9bf2c3..628f80f44c 100644 --- a/mindspore/ccsrc/frontend/parallel/graph_util/graph_splitter.cc +++ b/mindspore/ccsrc/frontend/parallel/graph_util/graph_splitter.cc @@ -51,7 +51,13 @@ ValueNodePtr CreateFakeValueNode(bool use_origin_node, const AnfNodePtr &origin_ tensor::TensorPtr fake_tensor = nullptr; if (use_origin_node) { MS_EXCEPTION_IF_NULL(origin_node); - auto origin_abstract = origin_node->abstract()->cast(); + abstract::AbstractTensorPtr origin_abstract; + if (origin_node->abstract()->isa()) { + auto get_one_tuple_element = origin_node->abstract()->cast()->elements()[0]; + origin_abstract = get_one_tuple_element->cast(); + } else { + origin_abstract = origin_node->abstract()->cast(); + } MS_EXCEPTION_IF_NULL(origin_abstract); fake_tensor = std::make_shared(origin_abstract->element()->BuildType()->type_id(), origin_abstract->shape()->shape()); @@ -186,11 +192,14 @@ CNodePtr CreateRecvNode(const FuncGraphPtr &func_graph, const InterProcessOpEdge } void ParameterServerMode::PreBuildDistributedGraph() { + MS_LOG(INFO) << "Start pre-building distribtued graph in Parameter Server mode."; MS_EXCEPTION_IF_NULL(node_labels_); ProcessForSplittedOptimizer(); + MS_LOG(INFO) << "End pre-building distribtued graph in Parameter Server mode."; } void ParameterServerMode::PostBuildDistributedGraph(const InterProcessOpEdgesInfo &comm_edges) { + MS_LOG(INFO) << "Start post-building distribtued graph in Parameter Server mode."; MS_EXCEPTION_IF_NULL(node_labels_); // Judge the node role number validation. uint32_t worker_num = ClusterContext::instance()->node_num(distributed::kEnvRoleOfWorker); @@ -225,7 +234,7 @@ void ParameterServerMode::PostBuildDistributedGraph(const InterProcessOpEdgesInf OperatorLabel worker_label = {i, distributed::kEnvRoleOfWorker}; InterProcessOpEdge edge = {ps_optimizer, node_labels_->at(ps_optimizer), dst_node, worker_label}; auto duplicated_send_node = CreateSendNode(func_graph_, edge); - node_labels_->at(duplicated_send_node) = edge.src_label; + node_labels_->insert(std::make_pair(duplicated_send_node, edge.src_label)); new_make_tuple_inputs.emplace_back(duplicated_send_node); } auto new_make_tuple_node = func_graph_->NewCNode(new_make_tuple_inputs); @@ -234,6 +243,7 @@ void ParameterServerMode::PostBuildDistributedGraph(const InterProcessOpEdgesInf } } } + MS_LOG(INFO) << "End post-building distribtued graph in Parameter Server mode."; } void ParameterServerMode::ProcessForSplittedOptimizer() { @@ -422,7 +432,11 @@ CNodePtr ParameterServerMode::CreateNodeWithInterProcessEdgeOnPServer(const std: MS_EXCEPTION_IF_NULL(new_node); // Step 3: Set the new node's abstract and attrs. - if (many_to_one_node_name == kConcatOpName) { + if (many_to_one_node_name == kAddNOpName) { + common::AnfAlgo::SetNodeAttr("N", MakeValue(static_cast(total_inputs_number)), new_node); + common::AnfAlgo::SetNodeAttr("n", MakeValue(static_cast(total_inputs_number)), new_node); + new_node->set_abstract(real_input->abstract()); + } else if (many_to_one_node_name == kConcatOpName) { auto origin_abs = real_input->abstract()->cast(); MS_EXCEPTION_IF_NULL(origin_abs); @@ -470,6 +484,7 @@ void GraphSplitter::Run() { if (std::find_if(node_labels_.begin(), node_labels_.end(), [&](const auto &node_to_label) { return node_to_label.second != this_process_label_; }) == node_labels_.end()) { + MS_LOG(INFO) << "No need to build and split distributed graph."; return; } @@ -498,7 +513,6 @@ void GraphSplitter::Run() { void GraphSplitter::DyeGraph() { MS_EXCEPTION_IF_NULL(func_graph_); - std::vector all_nodes = DeepScopedGraphSearch(func_graph_->get_return()); (void)std::for_each(all_nodes.begin(), all_nodes.end(), [this](const AnfNodePtr &node) { MS_EXCEPTION_IF_NULL(node); @@ -584,99 +598,22 @@ InterProcessOpEdgesInfo GraphSplitter::GenerateInterProcessOperators() { void GraphSplitter::SplitGraph(const std::vector &segments, const InterProcessOpEdgesInfo &comm_edges) { - // Traverse all the segments to add Depend for this process's graph. + // Step 1: Traverse all the segments to add Depend for this process's graph. // The list of corresponding in and out degrees. In another word, the map between one segments' input send nodes. and // output recv nodes. - std::vector, std::vector>> in_out_degree_list; - - // Traverse all the segments to add Depend for this process's graph. - for (const auto &segment : segments) { - // If this segment should be on current process, continue. - if (segment.label == this_process_label_) { - continue; - } - std::vector nodes = segment.nodes; - if (nodes.empty()) { - MS_LOG(EXCEPTION) << "This segment is empty."; - return; - } - - auto segment_first_node = nodes[0]; - if (node_labels_[segment_first_node] != segment.label) { - MS_LOG(EXCEPTION) << "Node label " << node_labels_[segment_first_node].to_string() - << " is not the same as segment label " << segment.label.to_string(); - } - - // Prepare for adding Depend between in-degree and out-degree of this segment because the execution order should be - // kept consistent. - std::vector concerned_in_degree_nodes = FindInterProcessInDegree(nodes, comm_edges); - std::vector concerned_out_degree_nodes = FindInterProcessOutDegree(nodes, comm_edges); - if (concerned_in_degree_nodes.empty()) { - continue; - } - in_out_degree_list.emplace_back(std::make_pair(concerned_in_degree_nodes, concerned_out_degree_nodes)); - } - + InOutDegreeList in_out_degree_list = GenerateInOutDegreeList(segments, comm_edges); if (in_out_degree_list.empty()) { - MS_LOG(ERROR) << "This process has no split graph. Optimize out the whole graph."; + MS_LOG(WARNING) << "After splitting, this process has no graph on it. So optimize out the whole graph."; auto return_value_node = CreateFakeValueNode(false); (void)func_graph_->manager()->Replace(func_graph_->output(), return_value_node); return; } - // This tuple is key to the dependency of send nodes so that they will not be optimized out in some cases. - std::vector send_node_tuple_inputs = {NewValueNode(prim::kPrimMakeTuple)}; - for (size_t i = 0; i < in_out_degree_list.size(); i++) { - auto &concerned_in_degree_nodes = in_out_degree_list[i].first; - auto &concerned_out_degree_nodes = in_out_degree_list[i].second; - send_node_tuple_inputs.insert(send_node_tuple_inputs.end(), concerned_in_degree_nodes.begin(), - concerned_in_degree_nodes.end()); - if (concerned_out_degree_nodes.empty()) { - // If this is the last segment's in and out degrees and has no out degrees, connect the send nodes to graph's - // output. - if (i == in_out_degree_list.size() - 1) { - auto make_tuple_node = func_graph_->NewCNode(send_node_tuple_inputs); - std::vector out = {NewValueNode(prim::kPrimDepend)}; - out.push_back(send_node_tuple_inputs.back()); - out.push_back(make_tuple_node); - auto out_node = func_graph_->NewCNode(out); - MS_EXCEPTION_IF_NULL(out_node); - out_node->set_abstract(send_node_tuple_inputs.back()->abstract()); - (void)func_graph_->manager()->Replace(func_graph_->output(), out_node); - } - } else { - auto make_tuple_node = func_graph_->NewCNode(send_node_tuple_inputs); - for (auto &recv : concerned_out_degree_nodes) { - std::vector depend_input = {NewValueNode(prim::kPrimDepend), recv->cast()->inputs()[1], - make_tuple_node}; - auto depend = func_graph_->NewCNode(depend_input); - depend->set_abstract(recv->cast()->inputs()[1]->abstract()); - func_graph_->manager()->SetEdge(recv, 1, depend); - } - // Reset the make tuple node inputs for next segments in degrees. - send_node_tuple_inputs = {NewValueNode(prim::kPrimMakeTuple)}; - } - } + // Step 2: Add dependency between segments on this process. + AddDependencyBetweenSegments(in_out_degree_list); - // Eliminate nodes which should be launched by other processes by set output edge. - for (auto &edge : comm_edges) { - InterProcessOpPair send_recv_pair = edge.second; - auto send_node = std::get<0>(send_recv_pair); - auto recv_node = std::get<1>(send_recv_pair); - auto user_node = std::get<2>(send_recv_pair); - int user_node_index = std::get<3>(send_recv_pair); - - OperatorLabel send_label = node_labels_[send_node]; - OperatorLabel recv_label = node_labels_[recv_node]; - if (send_label == recv_label) { - MS_LOG(EXCEPTION) << "The Send and Recv must have different label. But got Send: " << send_label.to_string() - << ", Recv: " << recv_label.to_string(); - } - - if (recv_label == this_process_label_) { - func_graph_->manager()->SetEdge(user_node, user_node_index, recv_node); - } - } + // Step 3: Eliminate nodes not on this process. + EliminateExtraNodes(comm_edges); } void GraphSplitter::DumpDistributedGraph(const InterProcessOpEdgesInfo &comm_edges) { @@ -792,6 +729,104 @@ std::vector GraphSplitter::FindInterProcessOutDegree(const std::vect return results; } +InOutDegreeList GraphSplitter::GenerateInOutDegreeList(const std::vector &segments, + const InterProcessOpEdgesInfo &comm_edges) { + MS_LOG(INFO) << "Start finding inter-process in-degrees."; + + InOutDegreeList in_out_degree_list; + // Traverse all the segments to add Depend for this process's graph. + for (const auto &segment : segments) { + // If this segment should be on current process, continue. + if (segment.label == this_process_label_) { + continue; + } + std::vector nodes = segment.nodes; + if (nodes.empty()) { + MS_LOG(EXCEPTION) << "This segment is empty."; + return in_out_degree_list; + } + + auto segment_first_node = nodes[0]; + if (node_labels_[segment_first_node] != segment.label) { + MS_LOG(EXCEPTION) << "Node label " << node_labels_[segment_first_node].to_string() + << " is not the same as segment label " << segment.label.to_string(); + } + + // Prepare for adding Depend between in-degree and out-degree of this segment because the execution order should be + // kept consistent. + std::vector concerned_in_degree_nodes = FindInterProcessInDegree(nodes, comm_edges); + std::vector concerned_out_degree_nodes = FindInterProcessOutDegree(nodes, comm_edges); + if (concerned_in_degree_nodes.empty()) { + continue; + } + in_out_degree_list.emplace_back(std::make_pair(concerned_in_degree_nodes, concerned_out_degree_nodes)); + } + MS_LOG(INFO) << "End finding inter-process in-degrees."; + return in_out_degree_list; +} + +void GraphSplitter::AddDependencyBetweenSegments(const InOutDegreeList &in_out_degree_list) { + MS_LOG(INFO) << "Start adding dependency between segments."; + // This tuple is key to the dependency of send nodes so that they will not be optimized out in some cases. + std::vector send_node_tuple_inputs = {NewValueNode(prim::kPrimMakeTuple)}; + for (size_t i = 0; i < in_out_degree_list.size(); i++) { + auto &concerned_in_degree_nodes = in_out_degree_list[i].first; + auto &concerned_out_degree_nodes = in_out_degree_list[i].second; + send_node_tuple_inputs.insert(send_node_tuple_inputs.end(), concerned_in_degree_nodes.begin(), + concerned_in_degree_nodes.end()); + if (concerned_out_degree_nodes.empty()) { + // If this is the last segment's in and out degrees and has no out degrees, connect the send nodes to graph's + // output. + if (i == in_out_degree_list.size() - 1) { + auto make_tuple_node = func_graph_->NewCNode(send_node_tuple_inputs); + std::vector out = {NewValueNode(prim::kPrimDepend)}; + out.push_back(send_node_tuple_inputs.back()); + out.push_back(make_tuple_node); + auto out_node = func_graph_->NewCNode(out); + MS_EXCEPTION_IF_NULL(out_node); + out_node->set_abstract(send_node_tuple_inputs.back()->abstract()); + (void)func_graph_->manager()->Replace(func_graph_->output(), out_node); + } + } else { + auto make_tuple_node = func_graph_->NewCNode(send_node_tuple_inputs); + for (auto &recv : concerned_out_degree_nodes) { + std::vector depend_input = {NewValueNode(prim::kPrimDepend), recv->cast()->inputs()[1], + make_tuple_node}; + auto depend = func_graph_->NewCNode(depend_input); + depend->set_abstract(recv->cast()->inputs()[1]->abstract()); + func_graph_->manager()->SetEdge(recv, 1, depend); + } + // Reset the make tuple node inputs for next segments in degrees. + send_node_tuple_inputs = {NewValueNode(prim::kPrimMakeTuple)}; + } + } + MS_LOG(INFO) << "End adding dependency between segments."; +} + +void GraphSplitter::EliminateExtraNodes(const InterProcessOpEdgesInfo &comm_edges) { + MS_LOG(INFO) << "Start eliminating nodes not on this process."; + // Eliminate nodes which should be launched by other processes by set output edge. + for (auto &edge : comm_edges) { + InterProcessOpPair send_recv_pair = edge.second; + auto send_node = std::get<0>(send_recv_pair); + auto recv_node = std::get<1>(send_recv_pair); + auto user_node = std::get<2>(send_recv_pair); + int user_node_index = std::get<3>(send_recv_pair); + + OperatorLabel send_label = node_labels_[send_node]; + OperatorLabel recv_label = node_labels_[recv_node]; + if (send_label == recv_label) { + MS_LOG(EXCEPTION) << "The Send and Recv must have different label. But got Send: " << send_label.to_string() + << ", Recv: " << recv_label.to_string(); + } + + if (recv_label == this_process_label_) { + func_graph_->manager()->SetEdge(user_node, user_node_index, recv_node); + } + } + MS_LOG(INFO) << "End eliminating nodes not on this process."; +} + bool GraphSplitter::IsNodesWithSameLabel(const AnfNodePtr &node1, const AnfNodePtr &node2) { if (node_labels_.count(node1) == 0 || node_labels_.count(node2) == 0) { MS_LOG(EXCEPTION) << "Either 'node1': " << node1->fullname_with_scope() diff --git a/mindspore/ccsrc/frontend/parallel/graph_util/graph_splitter.h b/mindspore/ccsrc/frontend/parallel/graph_util/graph_splitter.h index 34f6b0d142..bd88465723 100644 --- a/mindspore/ccsrc/frontend/parallel/graph_util/graph_splitter.h +++ b/mindspore/ccsrc/frontend/parallel/graph_util/graph_splitter.h @@ -104,6 +104,9 @@ struct InterProcessOpEdge { using InterProcessOpPair = std::tuple; using InterProcessOpEdgesInfo = std::map; +// The list of in and out degrees of one segment. +using InOutDegreeList = std::vector, std::vector>>; + constexpr char kAttrUpdateParameter[] = "update_parameter"; constexpr char kAttrParameterInputIndex[] = "parameter_input_index"; constexpr char kAttrGradientInputIndex[] = "gradient_input_index"; @@ -279,6 +282,16 @@ class GraphSplitter { std::vector FindInterProcessOutDegree(const std::vector &nodes, const InterProcessOpEdgesInfo &comm_edges); + // Generate in and out degrees list of the segments to add dependency between segments. + InOutDegreeList GenerateInOutDegreeList(const std::vector &segments, + const InterProcessOpEdgesInfo &comm_edges); + + // For the segments on this process, dependency edges should be created so that they won't be optimized out. + void AddDependencyBetweenSegments(const InOutDegreeList &in_out_degree_list); + + // Replace nodes inputs with Recv nodes to eliminate extra nodes not on this process. + void EliminateExtraNodes(const InterProcessOpEdgesInfo &comm_edges); + // Judge whether two nodes have the same distributed label. bool IsNodesWithSameLabel(const AnfNodePtr &node1, const AnfNodePtr &node2); diff --git a/mindspore/ccsrc/plugin/device/cpu/kernel/rpc/rpc_send_kernel.cc b/mindspore/ccsrc/plugin/device/cpu/kernel/rpc/rpc_send_kernel.cc index aa0a66ccc8..604ad727a4 100644 --- a/mindspore/ccsrc/plugin/device/cpu/kernel/rpc/rpc_send_kernel.cc +++ b/mindspore/ccsrc/plugin/device/cpu/kernel/rpc/rpc_send_kernel.cc @@ -20,7 +20,8 @@ namespace mindspore { namespace kernel { std::vector RpcSendKernelMod::GetOpSupport() { static std::vector support_list = { - KernelAttr().AddInputAttr(kNumberTypeFloat32).AddOutputAttr(kNumberTypeFloat32).AddAllSameAttr(true)}; + KernelAttr().AddInputAttr(kNumberTypeFloat32).AddOutputAttr(kNumberTypeFloat32).AddAllSameAttr(true), + KernelAttr().AddInputAttr(kNumberTypeInt32).AddOutputAttr(kNumberTypeInt32).AddAllSameAttr(true)}; return support_list; } diff --git a/mindspore/ccsrc/plugin/device/gpu/hal/device/gpu_comm_manager.cc b/mindspore/ccsrc/plugin/device/gpu/hal/device/gpu_comm_manager.cc index 2eaa399acb..9819dcf083 100644 --- a/mindspore/ccsrc/plugin/device/gpu/hal/device/gpu_comm_manager.cc +++ b/mindspore/ccsrc/plugin/device/gpu/hal/device/gpu_comm_manager.cc @@ -48,13 +48,13 @@ class GpuCommManager : public CommManager { bool GetRankID(const std::string &group, unsigned int *rank_id) const override { *rank_id = CollectiveInitializer::instance().GetRankIDByGroup(group); - MS_LOG(INFO) << "This process rank id is " << *rank_id << " in group " << group; + MS_LOG(DEBUG) << "This process rank id is " << *rank_id << " in group " << group; return true; } bool GetRankSize(const std::string &group, unsigned int *rank_size) const override { *rank_size = CollectiveInitializer::instance().GetGroupSize(group); - MS_LOG(INFO) << "Group " << group << " size is " << *rank_size; + MS_LOG(DEBUG) << "Group " << group << " size is " << *rank_size; return true; } diff --git a/mindspore/ccsrc/runtime/graph_scheduler/actor/rpc/recv_actor.cc b/mindspore/ccsrc/runtime/graph_scheduler/actor/rpc/recv_actor.cc index 58dfe46dae..10c150b549 100644 --- a/mindspore/ccsrc/runtime/graph_scheduler/actor/rpc/recv_actor.cc +++ b/mindspore/ccsrc/runtime/graph_scheduler/actor/rpc/recv_actor.cc @@ -26,6 +26,7 @@ namespace mindspore { namespace runtime { void RecvActor::SetOpcontext(OpContext *const op_context) { std::unique_lock lock(context_mtx_); + MS_EXCEPTION_IF_NULL(op_context); op_context_ = op_context; is_context_valid_ = true; context_cv_.notify_all(); @@ -71,6 +72,9 @@ bool RecvActor::StartServer() { } void RecvActor::RunOpInterProcessData(const std::shared_ptr &msg, OpContext *const context) { + // Once recv actor is launched, reset the op_context so that the next step's recv will not be launched in advance. + ResetOpcontext(); + MS_ERROR_IF_NULL_WO_RET_VAL(msg); MS_ERROR_IF_NULL_WO_RET_VAL(op_context_); auto &sequential_num = context->sequential_num_; @@ -119,12 +123,21 @@ bool RecvActor::CheckRunningCondition(const OpContext *context) co return true; } +void RecvActor::EraseInput(const OpContext *context) { + KernelActor::EraseInput(context); + if (input_op_inter_process_.count(context->sequential_num_) != 0) { + (void)input_op_inter_process_.erase(context->sequential_num_); + } +} + void RecvActor::HandleMessage(const std::shared_ptr &msg) { // Block the message handler if the context is invalid. std::unique_lock lock(context_mtx_); context_cv_.wait(lock, [this] { return is_context_valid_; }); lock.unlock(); + MS_LOG(INFO) << "Rpc actor recv message for inter-process edge: " << inter_process_edge_name_; + MS_ERROR_IF_NULL_WO_RET_VAL(msg); MS_ERROR_IF_NULL_WO_RET_VAL(op_context_); ActorDispatcher::Send(GetAID(), &RecvActor::RunOpInterProcessData, msg, op_context_); diff --git a/mindspore/ccsrc/runtime/graph_scheduler/actor/rpc/recv_actor.h b/mindspore/ccsrc/runtime/graph_scheduler/actor/rpc/recv_actor.h index ac715b3d1f..248eb89b70 100644 --- a/mindspore/ccsrc/runtime/graph_scheduler/actor/rpc/recv_actor.h +++ b/mindspore/ccsrc/runtime/graph_scheduler/actor/rpc/recv_actor.h @@ -35,7 +35,8 @@ class RecvActor : public RpcActor { GraphExecutionStrategy strategy, const std::set &modifiable_ref_input_indexes, const std::set &modifiable_ref_output_indexes) : RpcActor(name, kernel, device_context, memory_manager_aid, debug_aid, recorder_aid, strategy, - modifiable_ref_input_indexes, modifiable_ref_output_indexes, KernelTransformType::kRecvActor) {} + modifiable_ref_input_indexes, modifiable_ref_output_indexes, KernelTransformType::kRecvActor), + is_context_valid_(false) {} ~RecvActor() override = default; // Besides set the op context, this method also notify the message handler to 'RunOpInterProcessData'. @@ -53,13 +54,16 @@ class RecvActor : public RpcActor { bool StartServer(); protected: - // When an inter-process data received, this method is called. - void RunOpInterProcessData(const std::shared_ptr &msg, OpContext *const context); - // Besides the checking method in base class AbstractActor, condition of inter-process arrows should be checked for // recv actor. bool CheckRunningCondition(const OpContext *context) const override; + // When an inter-process data received, this method is called. + void RunOpInterProcessData(const std::shared_ptr &msg, OpContext *const context); + + // Besides erasing input data and input controls when finish actor running, inter-process inputs should be erased. + void EraseInput(const OpContext *context) override; + private: // The message callback of the tcp server. void HandleMessage(const std::shared_ptr &msg); diff --git a/mindspore/ccsrc/runtime/graph_scheduler/actor/rpc/rpc_actor.cc b/mindspore/ccsrc/runtime/graph_scheduler/actor/rpc/rpc_actor.cc index 11ebd76566..8ebb6888fc 100644 --- a/mindspore/ccsrc/runtime/graph_scheduler/actor/rpc/rpc_actor.cc +++ b/mindspore/ccsrc/runtime/graph_scheduler/actor/rpc/rpc_actor.cc @@ -18,12 +18,10 @@ namespace mindspore { namespace runtime { -void RpcActor::SetInterProcessEdgeName(const std::string &src_node_name, const std::string &dst_node_name) { - inter_process_edge_name_ = src_node_name + kInterProcessEdgeMark + dst_node_name; -} - void RpcActor::SetOpcontext(OpContext *const op_context) { op_context_ = op_context; } -void RpcActor::SetActorRouteRableProxy(const ActorRouteTableProxyPtr &proxy) { actor_route_table_proxy_ = proxy; } +void RpcActor::set_actor_route_table_proxy(const ActorRouteTableProxyPtr &proxy) { actor_route_table_proxy_ = proxy; } + +void RpcActor::set_inter_process_edge_name(const std::string &edge_name) { inter_process_edge_name_ = edge_name; } } // namespace runtime } // namespace mindspore diff --git a/mindspore/ccsrc/runtime/graph_scheduler/actor/rpc/rpc_actor.h b/mindspore/ccsrc/runtime/graph_scheduler/actor/rpc/rpc_actor.h index f2a772dc66..79cce2e056 100644 --- a/mindspore/ccsrc/runtime/graph_scheduler/actor/rpc/rpc_actor.h +++ b/mindspore/ccsrc/runtime/graph_scheduler/actor/rpc/rpc_actor.h @@ -63,10 +63,10 @@ class RpcActor : public KernelActor { virtual void ResetOpcontext() {} // Set the actor route proxy for rpc actors. - void SetActorRouteRableProxy(const ActorRouteTableProxyPtr &proxy); + void set_actor_route_table_proxy(const ActorRouteTableProxyPtr &proxy); // Set the inter-process edge name for rpc actor. - void SetInterProcessEdgeName(const std::string &src_node_name, const std::string &dst_node_name); + void set_inter_process_edge_name(const std::string &edge_name); // Set some info which will be used for rpc routing. virtual void SetRouteInfo(uint32_t peer_rank, const std::string &peer_role, const std::string &src_node_name, diff --git a/mindspore/ccsrc/runtime/graph_scheduler/actor/rpc/send_actor.cc b/mindspore/ccsrc/runtime/graph_scheduler/actor/rpc/send_actor.cc index 30e92f5e65..3961181a2a 100644 --- a/mindspore/ccsrc/runtime/graph_scheduler/actor/rpc/send_actor.cc +++ b/mindspore/ccsrc/runtime/graph_scheduler/actor/rpc/send_actor.cc @@ -22,7 +22,7 @@ namespace mindspore { namespace runtime { void SendActor::SetRouteInfo(uint32_t, const std::string &, const std::string &send_src_node_name, const std::string &send_dst_node_name) { - auto peer_actor_id = send_src_node_name + kInterProcessEdgeMark + send_dst_node_name; + auto peer_actor_id = inter_process_edge_name_; peer_actor_ids_.emplace_back(peer_actor_id); rpc_output_node_name_.emplace_back(send_dst_node_name); } @@ -69,6 +69,7 @@ void SendActor::SendOutput(OpContext *const context) { std::string peer_server_url = peer.second; auto message = BuildRpcMessage(send_output, peer_server_url); MS_ERROR_IF_NULL_WO_RET_VAL(message); + MS_LOG(INFO) << "Rpc actor send message for inter-process edge: " << peer.first; client_->SendAsync(std::move(message)); } } diff --git a/mindspore/ccsrc/runtime/graph_scheduler/rpc_node_scheduler.cc b/mindspore/ccsrc/runtime/graph_scheduler/rpc_node_scheduler.cc index c706dfe5f0..e2c2e6b040 100644 --- a/mindspore/ccsrc/runtime/graph_scheduler/rpc_node_scheduler.cc +++ b/mindspore/ccsrc/runtime/graph_scheduler/rpc_node_scheduler.cc @@ -48,7 +48,7 @@ RpcActorSetPtr RpcNodeScheduler::Build(const ActorSet *actor_set) { for (auto &rpc_actor : rpc_actors) { auto proxy = CreateRouteTableProxy(); MS_EXCEPTION_IF_NULL(proxy); - rpc_actor->SetActorRouteRableProxy(proxy); + rpc_actor->set_actor_route_table_proxy(proxy); } return rpc_actor_set; @@ -71,6 +71,7 @@ void RpcNodeScheduler::Link(const ActorSet *actor_set) { auto send_dst_roles = common::AnfAlgo::GetNodeAttr>(rpc_send_kernel, kAttrSendDstRoles); std::string send_src_node_name = common::AnfAlgo::GetNodeAttr(rpc_send_kernel, kAttrSendSrcNodeName); std::string send_dst_node_name = common::AnfAlgo::GetNodeAttr(rpc_send_kernel, kAttrSendDstNodeName); + std::string edge_name = common::AnfAlgo::GetNodeAttr(rpc_send_kernel, kAttrInterProcessEdgeName); if (send_dst_ranks.empty() || send_dst_roles.empty()) { MS_LOG(EXCEPTION) << "The attributes of send node " << rpc_send_kernel->fullname_with_scope() @@ -78,7 +79,7 @@ void RpcNodeScheduler::Link(const ActorSet *actor_set) { << ", send_src_node_name: " << send_src_node_name << ", send_dst_node_name: " << send_dst_node_name; } - send_actor->SetInterProcessEdgeName(send_src_node_name, send_dst_node_name); + send_actor->set_inter_process_edge_name(edge_name); send_actor->SetRouteInfo(send_dst_ranks[0], send_dst_roles[0], send_src_node_name, send_dst_node_name); } for (auto &recv_actor : recv_actors) { @@ -89,6 +90,7 @@ void RpcNodeScheduler::Link(const ActorSet *actor_set) { auto recv_src_roles = common::AnfAlgo::GetNodeAttr>(rpc_recv_kernel, kAttrRecvSrcRoles); std::string recv_src_node_name = common::AnfAlgo::GetNodeAttr(rpc_recv_kernel, kAttrRecvSrcNodeName); std::string recv_dst_node_name = common::AnfAlgo::GetNodeAttr(rpc_recv_kernel, kAttrRecvDstNodeName); + std::string edge_name = common::AnfAlgo::GetNodeAttr(rpc_recv_kernel, kAttrInterProcessEdgeName); if (recv_src_ranks.empty() || recv_src_roles.empty()) { MS_LOG(EXCEPTION) << "The attributes of recv node " << rpc_recv_kernel->fullname_with_scope() @@ -96,7 +98,7 @@ void RpcNodeScheduler::Link(const ActorSet *actor_set) { << ", recv_src_node_name: " << recv_src_node_name << ", recv_dst_node_name: " << recv_dst_node_name; } - recv_actor->SetInterProcessEdgeName(recv_src_node_name, recv_dst_node_name); + recv_actor->set_inter_process_edge_name(edge_name); recv_actor->SetRouteInfo(recv_src_ranks[0], recv_src_roles[0], recv_src_node_name, recv_dst_node_name); } }