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

Encapsulate methods and optimize logs.

This commit is contained in:
ZPaC 2022-03-29 15:17:13 +08:00
parent 693af7f260
commit 35e4e9aed6
10 changed files with 173 additions and 106 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

211
mindspore/ccsrc/frontend/parallel/graph_util/graph_splitter.cc
View File

@ -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>();
abstract::AbstractTensorPtr origin_abstract;
if (origin_node->abstract()->isa<abstract::AbstractTuple>()) {
auto get_one_tuple_element = origin_node->abstract()->cast<abstract::AbstractTuplePtr>()->elements()[0];
origin_abstract = get_one_tuple_element->cast<abstract::AbstractTensorPtr>();
} else {
origin_abstract = origin_node->abstract()->cast<abstract::AbstractTensorPtr>();
}
MS_EXCEPTION_IF_NULL(origin_abstract);
fake_tensor = std::make_shared<tensor::Tensor>(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<int64_t>(total_inputs_number)), new_node);
common::AnfAlgo::SetNodeAttr("n", MakeValue(static_cast<int64_t>(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<abstract::AbstractTensorPtr>();
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<AnfNodePtr> 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<SplitGraphSegment> &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::pair<std::vector<AnfNodePtr>, std::vector<AnfNodePtr>>> 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<AnfNodePtr> 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<AnfNodePtr> concerned_in_degree_nodes = FindInterProcessInDegree(nodes, comm_edges);
std::vector<AnfNodePtr> 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<AnfNodePtr> 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<AnfNodePtr> 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<AnfNodePtr> depend_input = {NewValueNode(prim::kPrimDepend), recv->cast<CNodePtr>()->inputs()[1],
make_tuple_node};
auto depend = func_graph_->NewCNode(depend_input);
depend->set_abstract(recv->cast<CNodePtr>()->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<AnfNodePtr> GraphSplitter::FindInterProcessOutDegree(const std::vect
return results;
}
InOutDegreeList GraphSplitter::GenerateInOutDegreeList(const std::vector<SplitGraphSegment> &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<AnfNodePtr> 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<AnfNodePtr> concerned_in_degree_nodes = FindInterProcessInDegree(nodes, comm_edges);
std::vector<AnfNodePtr> 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<AnfNodePtr> 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<AnfNodePtr> 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<AnfNodePtr> depend_input = {NewValueNode(prim::kPrimDepend), recv->cast<CNodePtr>()->inputs()[1],
make_tuple_node};
auto depend = func_graph_->NewCNode(depend_input);
depend->set_abstract(recv->cast<CNodePtr>()->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()

13
mindspore/ccsrc/frontend/parallel/graph_util/graph_splitter.h
View File

@ -104,6 +104,9 @@ struct InterProcessOpEdge {
using InterProcessOpPair = std::tuple<CNodePtr, CNodePtr, CNodePtr, int>;
using InterProcessOpEdgesInfo = std::map<InterProcessOpEdge, InterProcessOpPair>;
// The list of in and out degrees of one segment.
using InOutDegreeList = std::vector<std::pair<std::vector<AnfNodePtr>, std::vector<AnfNodePtr>>>;
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<AnfNodePtr> FindInterProcessOutDegree(const std::vector<AnfNodePtr> &nodes,
const InterProcessOpEdgesInfo &comm_edges);
// Generate in and out degrees list of the segments to add dependency between segments.
InOutDegreeList GenerateInOutDegreeList(const std::vector<SplitGraphSegment> &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);

3
mindspore/ccsrc/plugin/device/cpu/kernel/rpc/rpc_send_kernel.cc
View File

@ -20,7 +20,8 @@ namespace mindspore {
namespace kernel {
std::vector<KernelAttr> RpcSendKernelMod::GetOpSupport() {
static std::vector<KernelAttr> 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;
}

4
mindspore/ccsrc/plugin/device/gpu/hal/device/gpu_comm_manager.cc
View File

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

13
mindspore/ccsrc/runtime/graph_scheduler/actor/rpc/recv_actor.cc
View File

@ -26,6 +26,7 @@ namespace mindspore {
namespace runtime {
void RecvActor::SetOpcontext(OpContext<DeviceTensor> *const op_context) {
std::unique_lock<std::mutex> 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<MessageBase> &msg, OpContext<DeviceTensor> *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<DeviceTensor> *context) co
return true;
}
void RecvActor::EraseInput(const OpContext<DeviceTensor> *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<MessageBase> &msg) {
// Block the message handler if the context is invalid.
std::unique_lock<std::mutex> 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_);

12
mindspore/ccsrc/runtime/graph_scheduler/actor/rpc/recv_actor.h
View File

@ -35,7 +35,8 @@ class RecvActor : public RpcActor {
GraphExecutionStrategy strategy, const std::set<size_t> &modifiable_ref_input_indexes,
const std::set<size_t> &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<MessageBase> &msg, OpContext<DeviceTensor> *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<DeviceTensor> *context) const override;
// When an inter-process data received, this method is called.
void RunOpInterProcessData(const std::shared_ptr<MessageBase> &msg, OpContext<DeviceTensor> *const context);
// Besides erasing input data and input controls when finish actor running, inter-process inputs should be erased.
void EraseInput(const OpContext<DeviceTensor> *context) override;
private:
// The message callback of the tcp server.
void HandleMessage(const std::shared_ptr<MessageBase> &msg);

8
mindspore/ccsrc/runtime/graph_scheduler/actor/rpc/rpc_actor.cc
View File

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

4
mindspore/ccsrc/runtime/graph_scheduler/actor/rpc/rpc_actor.h
View File

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

3
mindspore/ccsrc/runtime/graph_scheduler/actor/rpc/send_actor.cc
View File

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

8
mindspore/ccsrc/runtime/graph_scheduler/rpc_node_scheduler.cc
View File

@ -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<std::vector<std::string>>(rpc_send_kernel, kAttrSendDstRoles);
std::string send_src_node_name = common::AnfAlgo::GetNodeAttr<std::string>(rpc_send_kernel, kAttrSendSrcNodeName);
std::string send_dst_node_name = common::AnfAlgo::GetNodeAttr<std::string>(rpc_send_kernel, kAttrSendDstNodeName);
std::string edge_name = common::AnfAlgo::GetNodeAttr<std::string>(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<std::vector<std::string>>(rpc_recv_kernel, kAttrRecvSrcRoles);
std::string recv_src_node_name = common::AnfAlgo::GetNodeAttr<std::string>(rpc_recv_kernel, kAttrRecvSrcNodeName);
std::string recv_dst_node_name = common::AnfAlgo::GetNodeAttr<std::string>(rpc_recv_kernel, kAttrRecvDstNodeName);
std::string edge_name = common::AnfAlgo::GetNodeAttr<std::string>(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);
}
}