/** * Copyright 2020 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. */ #include "graph/build/stream_allocator.h" #include #include #include "common/ge/ge_util.h" #include "framework/common/debug/ge_log.h" #include "framework/common/fmk_error_codes.h" #include "framework/common/types.h" #include "graph/build/logical_stream_allocator.h" #include "graph/common/omg_util.h" #include "graph/debug/ge_attr_define.h" #include "graph/ge_context.h" #include "graph/utils/graph_utils.h" #include "init/gelib.h" #include "common/string_util.h" #include "common/util/error_manager/error_manager.h" using std::map; using std::set; using std::string; using std::vector; namespace { const int64_t kTaskNumPerNormalNode = 3; const int64_t kTaskNumPerHcclNode = 245; const char *const kTrueStr = "true"; const char *const kFalseStr = "false"; const size_t kEventMultiplexingItemCount = 3; const size_t kKeyWordIndex = 0; const size_t kNodeNameIndex = 1; const size_t kEventIdIndex = 2; const char *const kSend = "SendTo"; const char *const kRecv = "RecvFrom"; const char kDelim = ';'; inline bool HasContinuousStreamLabel(const ge::OpDescPtr &op_desc, std::string &continuous_stream_label) { if (ge::AttrUtils::GetStr(op_desc, ge::ATTR_NAME_CONTINUOUS_STREAM_LABEL, continuous_stream_label)) { GELOGD("node[%s] get continuous_stream_label %s", op_desc->GetName().c_str(), continuous_stream_label.c_str()); return true; } return false; } bool IsHcclOp(const string &op_type) { const set hccl_op_types({ge::HCOMBROADCAST, ge::HCOMALLGATHER, ge::HCOMALLREDUCE, ge::HCOMREDUCESCATTER, ge::HCOMREDUCE}); return hccl_op_types.find(op_type) != hccl_op_types.end(); } ge::Status ParseNodeEventMultiplexing(const ge::NodePtr &node, const std::vector &raw_event_multiplexing, std::unordered_map>> &node_to_send, std::unordered_map>> &node_to_recv) { GE_CHECK_NOTNULL(node); for (const auto &str : raw_event_multiplexing) { std::vector ele = ge::StringUtils::Split(str, kDelim); if (ele.size() != kEventMultiplexingItemCount) { GELOGE(ge::PARAM_INVALID, "[Check][RawMultiplexing]Size error, node:%s, require size:%zu, actually:%zu.", node->GetName().c_str(), kEventMultiplexingItemCount, ele.size()); REPORT_INNER_ERROR("E19999", "Raw event multiplexing is invalid, node:%s, require size:%zu, actually:%zu.", node->GetName().c_str(), kEventMultiplexingItemCount, ele.size()); return ge::PARAM_INVALID; } int value; try { value = std::stoi(ele[kEventIdIndex]); } catch (std::invalid_argument &) { GELOGE(ge::PARAM_INVALID, "[Throw][Exception]Event id is invalid, node:%s, raw:%s.", node->GetName().c_str(), ele[kEventIdIndex].c_str()); REPORT_INNER_ERROR("E19999", "Event id is invalid, node:%s, raw:%s.", node->GetName().c_str(), ele[kEventIdIndex].c_str()); return ge::PARAM_INVALID; } catch (std::out_of_range &) { GELOGE(ge::PARAM_INVALID, "[Throw][Exception]Event id is out of range, node:%s, raw:%s.", node->GetName().c_str(), ele[kEventIdIndex].c_str()); REPORT_INNER_ERROR("E19999", "Event id is out of range, node:%s, raw:%s.", node->GetName().c_str(), ele[kEventIdIndex].c_str()); return ge::PARAM_INVALID; } if (value < 0) { GELOGE(ge::PARAM_INVALID, "[Check][EventId]Event id is out of range, node:%s, raw:%s, value:%d.", node->GetName().c_str(), ele[kEventIdIndex].c_str(), value); REPORT_INNER_ERROR("E19999", "Event id is out of range, node:%s, raw:%s, value:%d.", node->GetName().c_str(), ele[kEventIdIndex].c_str(), value); return ge::PARAM_INVALID; } if (ele[kKeyWordIndex] == kSend) { node_to_send[node].emplace_back(std::make_pair(ele[kNodeNameIndex], static_cast(value))); } else if (ele[kKeyWordIndex] == kRecv) { node_to_recv[node].emplace_back(std::make_pair(ele[kNodeNameIndex], static_cast(value))); } else { GELOGE(ge::PARAM_INVALID, "[Check][KeyWord]Key word is not supported, node:%s, key:%s.", node->GetName().c_str(), ele[kEventIdIndex].c_str()); REPORT_INNER_ERROR("E19999", "Key word is not supported, node:%s, key:%s.", node->GetName().c_str(), ele[kEventIdIndex].c_str()); return ge::PARAM_INVALID; } } return ge::SUCCESS; } ge::Status ParseAllNodeEventMultiplexing(const ge::ComputeGraphPtr &graph, std::unordered_map &name_to_node_map, std::unordered_map>> &node_to_send, std::unordered_map>> &node_to_recv) { for (const auto &node : graph->GetNodes(graph->GetGraphUnknownFlag())) { ge::OpDescPtr op_desc = node->GetOpDesc(); GE_CHECK_NOTNULL(op_desc); name_to_node_map.insert({node->GetName(), node}); std::vector raw_event_multiplexing; if (!(op_desc->HasAttr(ge::ATTR_NAME_EVENT_MULTIPLEXING))) { continue; } bool get_attr = ge::AttrUtils::GetListStr(op_desc, ge::ATTR_NAME_EVENT_MULTIPLEXING, raw_event_multiplexing); if (!get_attr) { GELOGE(ge::PARAM_INVALID, "[Get][Attr]Node:%s.", node->GetName().c_str()); REPORT_INNER_ERROR("E19999", "Failed to get raw event multiplexing, node:%s.", node->GetName().c_str()); return ge::PARAM_INVALID; } auto parse_ret = ParseNodeEventMultiplexing(node, raw_event_multiplexing, node_to_send, node_to_recv); if (parse_ret != ge::SUCCESS) { GELOGE(parse_ret, "[Parse][Eventmultiplexing]Node:%s.", node->GetName().c_str()); REPORT_INNER_ERROR("E19999", "Failed to parse node event multiplexing, node:%s.", node->GetName().c_str()); return parse_ret; } } return ge::SUCCESS; } std::vector GetIntersection(std::vector &a, std::vector &b) { std::unordered_set ele_of_a(a.begin(), a.end()); std::vector res; for (auto &ele : b) { if (ele_of_a.count(ele) > 0) { res.emplace_back(ele); } } return res; } } // namespace namespace ge { StreamAllocator::StreamAllocator(ComputeGraphPtr whole_graph, const Graph2SubGraphInfoList &subgraphs) : whole_graph_(std::move(whole_graph)), subgraphs_(subgraphs) { string single_stream_str; (void)GetContext().GetOption(ENABLE_SINGLE_STREAM, single_stream_str); const set stream_options = {"", kTrueStr, kFalseStr}; if (stream_options.find(single_stream_str) == stream_options.end()) { GELOGW("The value %s of the %s option is invalid, it should be true or false.", single_stream_str.c_str(), ENABLE_SINGLE_STREAM); } enable_single_stream_ = (single_stream_str == kTrueStr) ? true : false; GELOGD("Enable single stream: %s.", enable_single_stream_ ? kTrueStr : kFalseStr); } Status StreamAllocator::AssignLogicalStreams(const std::map &max_parallel_num, bool hcom_parallel) { GE_CHECK_NOTNULL(whole_graph_); GE_DUMP(whole_graph_, "BeforeAssignedLogicalStreams"); auto gelib = GELib::GetInstance(); if (gelib == nullptr) { REPORT_INNER_ERROR("E19999", "Check GELib instance nullptr"); GELOGE(FAILED, "Get GELib instance failed."); return FAILED; } const map &scheduler_confs = gelib->DNNEngineManagerObj().GetSchedulers(); LogicalStreamAllocator logical_allocator(scheduler_confs, max_parallel_num); logical_allocator.EnableSingleStream(enable_single_stream_); logical_allocator.EnableHcomParallel(hcom_parallel); Status status = logical_allocator.Assign(whole_graph_, subgraphs_, stream_num_); if (status != SUCCESS) { GELOGE(status, "Assign logical streams failed."); return status; } GE_DUMP(whole_graph_, "AfterAssignedLogicalStreams"); return SUCCESS; } // After allocating the logical stream in the graph, refresh the stream in the // graph and insert the synchronization node. Status StreamAllocator::RefreshRealStream(int64_t &stream_num, int64_t &event_num) { GE_CHECK_NOTNULL(whole_graph_); GE_DUMP(whole_graph_, "BeforeRefreshRealStream"); Status status = AssignSingleStream(); if (status != SUCCESS) { GELOGE(status, "AssignSingleStream failed!"); return status; } status = SetActiveStreamsByLabel(); if (status != SUCCESS) { GELOGE(status, "SetActiveStreamsByLabel failed!"); return status; } status = SetActiveStreamsForSubgraphs(); if (status != SUCCESS) { GELOGE(status, "SetActiveStreamsForSubgraphs failed."); return status; } status = InsertSyncEvents(); if (status != SUCCESS) { GELOGE(status, "InsertSyncEventId failed!"); return status; } status = OptimizeSyncEvents(); if (status != SUCCESS) { GELOGE(status, "OptimizeSyncEventId failed!"); return status; } vector> split_streams(stream_num_); status = SplitStreams(split_streams); if (status != SUCCESS) { GELOGE(status, "SplitStreams failed!"); return status; } status = UpdateActiveStreams(split_streams); if (status != SUCCESS) { GELOGE(status, "UpdateActiveStreams failed!"); return status; } status = RefreshContinuousEvents(); if (status != SUCCESS) { GELOGE(status, "RefreshContinuousEvents failed!"); return status; } status = RefreshEventsWithReuse(); if (status != SUCCESS) { GELOGE(status, "[Refresh][Events]RefreshEventsWithReuse failed!"); return status; } status = InsertSyncEventNodes(); if (status != SUCCESS) { GELOGE(status, "InsertSyncEventNode failed!"); return status; } DumpEvents(); GE_DUMP(whole_graph_, "AfterRefreshRealStream"); for (const NodePtr &node : whole_graph_->GetNodes(whole_graph_->GetGraphUnknownFlag())) { GE_CHECK_NOTNULL(node->GetOpDesc()); auto stream_id = node->GetOpDesc()->GetStreamId(); if (stream_id == kInvalidStream) { node->GetOpDesc()->SetStreamId(0); } } if (stream_num_ == 0) { GELOGI("None of nodes need to assign stream, stream num is 0, it will cause error, so change it to 1"); stream_num_ = 1; } GELOGD("stream num: %ld, event num: %u.", stream_num_, event_num_); stream_num = stream_num_; event_num = static_cast(event_num_); return SUCCESS; } Status StreamAllocator::AssignSingleStream() { if (!enable_single_stream_) { return SUCCESS; } if (stream_num_ > 1) { REPORT_INNER_ERROR("E19999", "The number of ts streams is %ld, only one is supported", stream_num_); GELOGE(FAILED, "The number of ts streams is %ld, only one is supported.", stream_num_); return FAILED; } int64_t task_count = 0; for (const NodePtr &node : whole_graph_->GetNodes(whole_graph_->GetGraphUnknownFlag())) { string op_type = node->GetType(); if (IsHcclOp(op_type)) { task_count += kTaskNumPerHcclNode; } else { task_count += kTaskNumPerNormalNode; } } uint32_t max_normal_stream_count = 0; uint32_t max_normal_task_count = 0; Status status = GetMaxStreamAndTask(false, max_normal_stream_count, max_normal_task_count); if (status != SUCCESS) { GELOGE(status, "Get max task count of normal stream failed."); return status; } if (task_count > static_cast(max_normal_task_count)) { uint32_t max_huge_stream_count = 0; uint32_t max_huge_task_count = 0; Status status = GetMaxStreamAndTask(true, max_huge_stream_count, max_huge_task_count); if (status == SUCCESS) { int64_t huge_stream = 0; GELOGI("Use huge stream %ld.", huge_stream); huge_streams_.emplace_back(huge_stream); } else { GELOGW( "The estimated task count %ld is greater than the max count of normal stream," " but the huge stream is not supported.", task_count); } } return SUCCESS; } Status StreamAllocator::SetActiveStreamsByLabel() { for (const auto &node : whole_graph_->GetNodes(whole_graph_->GetGraphUnknownFlag())) { OpDescPtr op_desc = node->GetOpDesc(); GE_CHECK_NOTNULL(op_desc); string stream_label; if (AttrUtils::GetStr(op_desc, ATTR_NAME_STREAM_LABEL, stream_label) && !stream_label.empty()) { int64_t stream_id = op_desc->GetStreamId(); if (stream_id != kInvalidStream) { labeled_streams_[stream_label].emplace(stream_id); } } } for (const auto &node : whole_graph_->GetNodes(whole_graph_->GetGraphUnknownFlag())) { GE_CHECK_NOTNULL(node->GetOpDesc()); vector activated_label_list; if (!AttrUtils::GetListStr(node->GetOpDesc(), ATTR_NAME_ACTIVE_LABEL_LIST, activated_label_list) || activated_label_list.empty()) { continue; } vector activated_stream_list; for (string &activated_label : activated_label_list) { specific_activated_labels_[activated_label].emplace(node); for (int64_t activated_stream : labeled_streams_[activated_label]) { activated_stream_list.push_back(static_cast(activated_stream)); specific_activated_streams_.emplace(activated_stream); specific_activated_streams_nodes_map_[activated_stream].emplace(node); GELOGI("Node %s active stream %ld by %s.", node->GetName().c_str(), activated_stream, activated_label.c_str()); } } GE_CHK_BOOL_EXEC(AttrUtils::SetListInt(node->GetOpDesc(), ATTR_NAME_ACTIVE_STREAM_LIST, activated_stream_list), REPORT_INNER_ERROR("E19999", "Set Attr:%s for op:%s(%s) failed", ATTR_NAME_ACTIVE_STREAM_LIST.c_str(), node->GetName().c_str(), node->GetType().c_str()); GELOGE(FAILED, "SetListInt failed."); return FAILED); } return SUCCESS; } Status StreamAllocator::SetActiveStreamsForSubgraphs() { for (auto &subgraph : whole_graph_->GetAllSubgraphs()) { GE_CHECK_NOTNULL(subgraph); NodePtr first_active_node = nullptr; // Get all streams in subgraph. set subgraph_streams; for (auto &node : subgraph->GetDirectNode()) { OpDescPtr op_desc = node->GetOpDesc(); GE_CHECK_NOTNULL(op_desc); // Skip streams with label string stream_label; if (AttrUtils::GetStr(op_desc, ATTR_NAME_STREAM_LABEL, stream_label) && !stream_label.empty()) { continue; } int64_t stream_id = op_desc->GetStreamId(); if (stream_id != kInvalidStream) { subgraph_streams.emplace(stream_id); GELOGI("Add stream %ld to active_stream_list of node %s of graph %s", stream_id, node->GetName().c_str(), subgraph->GetName().c_str()); } bool is_first_active = false; if (AttrUtils::GetBool(op_desc, ATTR_NAME_SUBGRAPH_FIRST_ACTIVE, is_first_active) && is_first_active) { first_active_node = node; } } if (first_active_node == nullptr) { continue; } subgraph_first_active_node_map_[subgraph] = first_active_node; // Set active streams for StreamActive. subgraph_streams.erase(first_active_node->GetOpDesc()->GetStreamId()); vector active_streams; for (int64_t active_stream : subgraph_streams) { active_streams.emplace_back(static_cast(active_stream)); specific_activated_streams_.emplace(active_stream); } if (!AttrUtils::SetListInt(first_active_node->GetOpDesc(), ATTR_NAME_ACTIVE_STREAM_LIST, active_streams)) { REPORT_INNER_ERROR("E19999", "Set Attr:%s for op:%s(%s) failed", ATTR_NAME_ACTIVE_STREAM_LIST.c_str(), first_active_node->GetName().c_str(), first_active_node->GetType().c_str()); GELOGE(FAILED, "Set active streams for node %s failed.", first_active_node->GetName().c_str()); return FAILED; } } return SUCCESS; } // Insert the send/recv event id to the graph Status StreamAllocator::InsertSyncEvents() { for (const auto &cur_node : whole_graph_->GetNodes(whole_graph_->GetGraphUnknownFlag())) { // Take the adjacent points, then judge whether need to insert the event for (const OutDataAnchorPtr &anchor : cur_node->GetAllOutDataAnchors()) { for (const InDataAnchorPtr &peer_in_anchor : anchor->GetPeerInDataAnchors()) { NodePtr next_node = peer_in_anchor->GetOwnerNode(); Status status = InsertOneEventInTwoNodes(cur_node, next_node); if (status != SUCCESS) { GELOGE(status, "InsertOneEventInTwoNodes failed!"); return status; } } } /// If the two nodes of the control side belong to two streams, /// you also need to add the send/recv event. if (cur_node->GetOutControlAnchor() != nullptr) { for (const AnchorPtr &peer_in_anchor : cur_node->GetOutControlAnchor()->GetPeerAnchors()) { NodePtr next_node = peer_in_anchor->GetOwnerNode(); Status status = InsertOneEventInTwoNodes(cur_node, next_node); if (status != SUCCESS) { GELOGE(status, "InsertOneEventInTwoNodes failed!"); return status; } } } } Status status = InsertEventsForSubgraph(); if (status != SUCCESS) { GELOGE(status, "InsertEventsBetweenSubAndParentGraphNodes failed!"); return status; } return SUCCESS; } // Insert one send/recv event in two nodes Status StreamAllocator::InsertOneEventInTwoNodes(const NodePtr &cur_node, const NodePtr &next_node) { GE_CHECK_NOTNULL(cur_node->GetOpDesc()); GE_CHECK_NOTNULL(next_node->GetOpDesc()); // No need to insert events after node that do not assign streams. int64_t cur_stream_id = cur_node->GetOpDesc()->GetStreamId(); if (cur_stream_id == kInvalidStream) { GELOGD("No need to insert event after node %s.", cur_node->GetName().c_str()); return SUCCESS; } // No need to insert events between nodes in the same stream. int64_t next_stream_id = next_node->GetOpDesc()->GetStreamId(); if (cur_stream_id == next_stream_id) { return SUCCESS; } if (((cur_node->GetType() == ENTER) || (cur_node->GetType() == REFENTER)) && (next_node->GetType() != STREAMACTIVE)) { GELOGD("No need to insert event between %s and %s.", cur_node->GetName().c_str(), next_node->GetName().c_str()); return SUCCESS; } if (next_stream_id == kInvalidStream) { REPORT_INNER_ERROR("E19999", "Stream id of next_node %s(%s) should not be %ld", next_node->GetName().c_str(), next_node->GetType().c_str(), kInvalidStream); GELOGE(FAILED, "Stream id of next_node %s should not be %ld", next_node->GetName().c_str(), kInvalidStream); return FAILED; } // No event needs to be inserted between the active node and the activated stream. string next_node_label; if (AttrUtils::GetStr(next_node->GetOpDesc(), ATTR_NAME_STREAM_LABEL, next_node_label) && !next_node_label.empty()) { auto iter = specific_activated_labels_.find(next_node_label); if (iter != specific_activated_labels_.end()) { for (const auto &active_node : iter->second) { OpDescPtr active_op = active_node->GetOpDesc(); GE_CHECK_NOTNULL(active_op); if ((cur_stream_id == active_op->GetStreamId()) && (cur_node->GetOpDesc()->GetId() <= active_op->GetId())) { GELOGI("No need to insert event between node %s and %s.", cur_node->GetName().c_str(), next_node->GetName().c_str()); return SUCCESS; } } } } // Add send and receive events. AddSendEventId(cur_node, event_num_); AddRecvEventId(next_node, event_num_); GELOGD("Insert event %u between node %s(stream %ld) and %s(stream %ld)", event_num_, cur_node->GetName().c_str(), cur_stream_id, next_node->GetName().c_str(), next_stream_id); ++event_num_; return SUCCESS; } Status StreamAllocator::InsertEventsForSubgraph() { for (const auto &subgraph : whole_graph_->GetAllSubgraphs()) { GE_CHECK_NOTNULL(subgraph); for (const auto &node : subgraph->GetDirectNode()) { auto op_desc = node->GetOpDesc(); GE_CHECK_NOTNULL(op_desc); bool is_subgraph_end_node = false; if (!AttrUtils::GetBool(op_desc, ATTR_NAME_SUBGRAPH_END_NODE, is_subgraph_end_node) || !is_subgraph_end_node) { continue; } const auto parent_node = subgraph->GetParentNode(); GE_CHECK_NOTNULL(parent_node); // Insert events between subgraph end node and parent node's out nodes for (const auto &next_node : parent_node->GetOutAllNodes()) { Status status = InsertOneEventInTwoNodes(node, next_node); if (status != SUCCESS) { GELOGE(status, "InsertOneEventInTwoNodes failed!"); return status; } } break; } } return SUCCESS; } // Optimize the event in the graph, delete the redundant sync event according to the stream information Status StreamAllocator::OptimizeSyncEvents() { map> stream_nodes; for (const auto &node : whole_graph_->GetNodes(whole_graph_->GetGraphUnknownFlag())) { GE_CHECK_NOTNULL(node->GetOpDesc()); int64_t stream_id = node->GetOpDesc()->GetStreamId(); stream_nodes[stream_id].emplace_back(node); } Status status = OptimizeBySendEvents(stream_nodes); if (status != SUCCESS) { GELOGE(status, "OptimizeBySendEvents failed!"); return status; } status = OptimizeByRecvEvents(stream_nodes); if (status != SUCCESS) { GELOGE(status, "OptimizeByRecvEvents failed!"); return status; } status = OptimizeByStreamActivate(); if (status != SUCCESS) { GELOGE(status, "OptimizeByStreamActivate failed!"); return status; } for (auto pair : node_to_send_events_) { if (pair.first->GetType() == STREAMSWITCH) { for (auto event_id : pair.second) { GELOGI("Curren switch node is %s, remove send event_id %d.", pair.first->GetName().c_str(), event_id); RmvSendEventId(pair.first, event_id); auto recv_node = GetNodeFromRecvEventId(event_id); GELOGI("Curren recv_node is %s, remove recv event_id %d.", recv_node->GetName().c_str(), event_id); RmvRecvEventId(recv_node, event_id); } } } return SUCCESS; } /// Optimization scenario: one stream has multiple send events in one node, /// and multiple nodes for recv events on another stream /// Example: /// Stream0 Stream1 /// N1 - - - event - > N1 /// \ | /// \ v /// - - event - > N2 Status StreamAllocator::OptimizeBySendEvents(const map> &stream_nodes) { for (const auto &one_pair : stream_nodes) { // The nodes on a stream in order const vector &nodes = one_pair.second; map send_node_to_event_id; for (const auto &recv_node_ptr : nodes) { GE_CHECK_NOTNULL(recv_node_ptr); // Get all recv events of the current node, then traverse the event vector recv_events; GetRecvEventIdList(recv_node_ptr, recv_events); for (const auto &event_id : recv_events) { NodePtr send_node_ptr = GetNodeFromSendEventId(event_id); GE_CHECK_NOTNULL(send_node_ptr); /// If the record to the stream is found in the map, /// and the recv node is the node, then remove sync event if (send_node_to_event_id.find(send_node_ptr) != send_node_to_event_id.end()) { RmvSendEventId(send_node_ptr, event_id); RmvRecvEventId(recv_node_ptr, event_id); GELOGI("Remove event %u between node %s and node %s", event_id, send_node_ptr->GetName().c_str(), recv_node_ptr->GetName().c_str()); } else { send_node_to_event_id[send_node_ptr] = event_id; } } } } return SUCCESS; } /// Scenario: multiple send nodes on a stream sent to a single recv node on the destination stream /// Example: /// Stream0 Stream1 /// N1 - - /// | | /// | - - event - - - /// | | /// V V /// N2 - - - event - > N2 Status StreamAllocator::OptimizeByRecvEvents(const map> &stream_nodes) { for (const auto &one_pair : stream_nodes) { // The nodes on a stream in order const vector &nodes = one_pair.second; map recv_node_to_event_id; for (const auto &send_node_ptr : nodes) { GE_CHECK_NOTNULL(send_node_ptr); // Get all send events of the current node, then traverse the event vector send_id_list; GetSendEventIdList(send_node_ptr, send_id_list); for (const auto &event_id : send_id_list) { NodePtr recv_node_ptr = GetNodeFromRecvEventId(event_id); GE_CHECK_NOTNULL(recv_node_ptr); /// If the record to the stream is found in the map, /// and the send node is the node, then remove sync event auto it = recv_node_to_event_id.find(recv_node_ptr); if (it != recv_node_to_event_id.end()) { uint32_t pre_event_id = it->second; NodePtr pre_send_node_ptr = GetNodeFromSendEventId(pre_event_id); GE_CHECK_NOTNULL(pre_send_node_ptr); RmvSendEventId(pre_send_node_ptr, pre_event_id); RmvRecvEventId(recv_node_ptr, pre_event_id); GELOGI("Remove event %u between node %s and node %s.", event_id, pre_send_node_ptr->GetName().c_str(), recv_node_ptr->GetName().c_str()); } recv_node_to_event_id[recv_node_ptr] = event_id; } } } return SUCCESS; } Status StreamAllocator::OptimizeByStreamActivate() { auto node_to_send_events_temp = node_to_send_events_; for (const auto &node_event_id_pair : node_to_send_events_temp) { const NodePtr &send_node_ptr = node_event_id_pair.first; for (const auto &event_id : node_event_id_pair.second) { NodePtr recv_node_ptr = GetNodeFromRecvEventId(event_id); GE_CHECK_NOTNULL(recv_node_ptr); if (IsRecvNodeActivatedBySendNode(send_node_ptr, recv_node_ptr)) { RmvSendEventId(send_node_ptr, event_id); RmvRecvEventId(recv_node_ptr, event_id); GELOGI("Remove event %u between node %s and node %s.", event_id, send_node_ptr->GetName().c_str(), recv_node_ptr->GetName().c_str()); } } } return SUCCESS; } // In situation : stream(normal) -> stream(streamActivate)-> // -> stream(streamSwitch) -> stream(streamActivate) -> stream(stream true or false) // No need to insert an event between node in stream(normal) and node in stream(stream true or false) bool StreamAllocator::IsRecvNodeActivatedBySendNode(const NodePtr &send_node_ptr, const NodePtr &recv_node_ptr) const { GE_CHECK_NOTNULL_EXEC(send_node_ptr->GetOpDesc(), REPORT_INNER_ERROR("E19999", "Check param send_node_ptr nullptr"); GELOGE(FAILED, "op desc is nullptr"); return false); GE_CHECK_NOTNULL_EXEC(recv_node_ptr->GetOpDesc(), REPORT_INNER_ERROR("E19999", "Check param recv_node_ptr nullptr"); GELOGE(FAILED, "op desc is nullptr"); return false); auto cur_stream_id = send_node_ptr->GetOpDesc()->GetStreamId(); if (AttrUtils::HasAttr(recv_node_ptr->GetOpDesc(), ATTR_NAME_STREAM_LABEL)) { // find streamActivate node auto iter = specific_activated_streams_nodes_map_.find(recv_node_ptr->GetOpDesc()->GetStreamId()); set activate_stream_nodes; if (iter != specific_activated_streams_nodes_map_.end()) { activate_stream_nodes = iter->second; } set visited_nodes{recv_node_ptr}; while (!activate_stream_nodes.empty()) { set activate_stream_nodes_temp; for (const auto &activate_stream_node : activate_stream_nodes) { GE_IF_BOOL_EXEC(activate_stream_node->GetOpDesc() == nullptr, continue); if (visited_nodes.find(activate_stream_node) != visited_nodes.end() || AttrUtils::HasAttr(activate_stream_node->GetOpDesc(), ATTR_NAME_IS_LOOP_ACTIVE)) { return false; } /// /// stream_0 --> stream_2 --> stream_3 --> stream_4 /// /\ | /// | \/ /// | stream_1 --> stream_5 --> stream_6 --> stream_7 /// | /\ | | /// | | \/ | /// | |---------- stream_8 | /// | | /// |-----------------------------------------------------------| /// /// Exit1(S7) Exit2(S7) Exit3(S7) /// \ / | /// AddN(S1) NextIteration(S7) /// | | /// NextIteration(S1) / /// | / /// | / /// StreamActive(S7) /// /// Event between Exit1/Exit2 and AddN should not be optimized /// if (IsActiveAfterNextIteration(activate_stream_node)) { continue; } visited_nodes.insert(activate_stream_node); // nodes in stream link to streamActivate no need to add event before activated node for (const auto &pre_activate_stream_node : activate_stream_node->GetInNodes()) { GE_IF_BOOL_EXEC(pre_activate_stream_node->GetOpDesc() == nullptr, continue); if (pre_activate_stream_node->GetOpDesc()->GetStreamId() == cur_stream_id && pre_activate_stream_node->GetOpDesc()->GetId() >= send_node_ptr->GetOpDesc()->GetId()) { return true; } auto in_nodes_of_pre = pre_activate_stream_node->GetInNodes(); if (std::find(in_nodes_of_pre.begin(), in_nodes_of_pre.end(), send_node_ptr) != in_nodes_of_pre.end()) { return true; } } auto iterator = specific_activated_streams_nodes_map_.find(activate_stream_node->GetOpDesc()->GetStreamId()); if (iterator != specific_activated_streams_nodes_map_.end()) { auto active_nodes = iterator->second; for (const auto &active_node : active_nodes) { activate_stream_nodes_temp.emplace(active_node); } } } activate_stream_nodes = activate_stream_nodes_temp; } } return false; } bool StreamAllocator::IsActiveAfterNextIteration(const NodePtr &active_node_ptr) const { if ((active_node_ptr == nullptr) || active_node_ptr->GetInControlNodes().empty()) { return false; } for (const auto &in_node : active_node_ptr->GetInControlNodes()) { if ((in_node->GetType() != NEXTITERATION) && (in_node->GetType() != REFNEXTITERATION)) { return false; } } return true; } // Split the stream according to the maximum number of nodes in the stream. Status StreamAllocator::SplitStreams(vector> &split_streams) { if (enable_single_stream_ || stream_num_ == 0) { GELOGI("The single stream option is enabled or the number of streams is 0, no need to split streams."); return SUCCESS; } // stream_node_num_vec records the number of all nodes on each stream // added_stream_num_vec records the number of streams that each stream needs to increase // new_stream_id_vec records the new physical stream id for each stream vector stream_node_num_vec(stream_num_); vector added_stream_num_vec(stream_num_); vector new_stream_id_vec(stream_num_); map stream_continuous_2_node_num_map; map> stream_continuous_2_nodes_map; map> stream_2_nodes_map; vector pre_node_vec(stream_num_); int64_t last_stream_id = stream_num_ - 1; for (auto i = 0; i <= last_stream_id; i++) { stream_node_num_vec[i] = 0; added_stream_num_vec[i] = 0; new_stream_id_vec[i] = i; pre_node_vec[i] = nullptr; } uint32_t max_stream_count = 0; uint32_t max_task_count = 0; GE_CHK_STATUS_RET(GetMaxStreamAndTask(false, max_stream_count, max_task_count), "Get max stream and task count failed."); for (const auto &cur_node : whole_graph_->GetNodes(whole_graph_->GetGraphUnknownFlag())) { GE_CHECK_NOTNULL(cur_node); auto op_desc = cur_node->GetOpDesc(); GE_CHECK_NOTNULL(op_desc); int64_t stream_id = op_desc->GetStreamId(); if (stream_id == kInvalidStream) { continue; } if (stream_id > last_stream_id) { REPORT_INNER_ERROR("E19999", "streamid(%ld) > last_stream_id(%ld), check invalid", stream_id, last_stream_id); GELOGE(FAILED, "SplitStreams:streamid(%ld) > last_stream_id(%ld)", stream_id, last_stream_id); return FAILED; } bool is_stream_first_node = (stream_node_num_vec[stream_id] == 0); AddNodeNum(cur_node, stream_node_num_vec[stream_id]); stream_2_nodes_map[stream_id].push_back(cur_node); // The maximum number of tasks per stream. int64_t max_node_num_one_stream = GetMaxNodeNumPerStream(cur_node, max_task_count); std::string continuous_stream_label; if (HasContinuousStreamLabel(op_desc, continuous_stream_label)) { stream_continuous_2_node_num_map[continuous_stream_label]++; // return error if (stream_continuous_2_node_num_map[continuous_stream_label] > max_node_num_one_stream) { REPORT_INNER_ERROR("E19999", "Check node[%s] stream_id[%ld] continuous stream label[%s] unsatisfied", op_desc->GetName().c_str(), stream_id, continuous_stream_label.c_str()); GELOGE(FAILED, "SplitStreams:node[%s] stream_id[%ld] continuous stream label[%s] unsatisfied ", op_desc->GetName().c_str(), stream_id, continuous_stream_label.c_str()); return FAILED; } stream_continuous_2_nodes_map[continuous_stream_label].push_back(cur_node); } // Split the stream if it exceeds the maximum number of nodes in the stream. if (NeedSpiltNewStream(stream_node_num_vec[stream_id], max_node_num_one_stream, op_desc, is_stream_first_node)) { last_stream_id++; GELOGI( "stream_node_num_vec[%ld]= %ld > max_node_num_one_stream : %ld, " "It's time to split the stream, split newly-added stream id is %ld", stream_id, stream_node_num_vec[stream_id], max_node_num_one_stream, last_stream_id); NodePtr pre_node = pre_node_vec[stream_id]; stream_node_num_vec[stream_id] = 0; AddNodeNum(cur_node, stream_node_num_vec[stream_id]); // try spilt a new stream and move same continuous stream label nodes from this stream bool not_use_cur = false; NodePtr not_cur = nullptr; std::string cur_continuous_stream_label; if (HasContinuousStreamLabel(op_desc, cur_continuous_stream_label)) { // get stored nodes auto nodes = stream_continuous_2_nodes_map[cur_continuous_stream_label]; GE_RETURN_WITH_LOG_IF_FALSE(!nodes.empty(), "split stream with continuous stream label %s failed", cur_continuous_stream_label.c_str()); for (const auto &node : nodes) { auto stored_op_desc = node->GetOpDesc(); GE_CHECK_NOTNULL(stored_op_desc); stored_op_desc->SetStreamId(last_stream_id); AddNodeNum(node, stream_node_num_vec[stream_id]); } not_use_cur = true; not_cur = nodes.front(); GE_CHECK_NOTNULL(not_cur); GELOGI("split from first node %s with continuous stream label %s", not_cur->GetName().c_str(), cur_continuous_stream_label.c_str()); auto iter = std::find(stream_2_nodes_map[stream_id].begin(), stream_2_nodes_map[stream_id].end(), not_cur); GE_RETURN_WITH_LOG_IF_FALSE( (iter != stream_2_nodes_map[stream_id].end()) && (iter != stream_2_nodes_map[stream_id].begin()), "split stream with continuous stream label %s failed", cur_continuous_stream_label.c_str()); iter--; pre_node = *iter; } added_stream_num_vec[stream_id]++; new_stream_id_vec[stream_id] = last_stream_id; split_streams[stream_id].emplace(last_stream_id); node_split_stream_map_[cur_node] = last_stream_id; // Add the send/recv event to the first and last nodes of the split stream. if (pre_node != nullptr) { GE_CHK_STATUS_RET(AddEventId(pre_node, not_cur, cur_node, not_use_cur), "AddEventId failed."); } } /// If the split stream num is greater than 1, the node behind the same /// stream must reset the new stream id. if (added_stream_num_vec[stream_id] >= 1) { op_desc->SetStreamId(new_stream_id_vec[stream_id]); } pre_node_vec[stream_id] = cur_node; } if (last_stream_id >= 0) { stream_num_ = last_stream_id + 1; } return SUCCESS; } bool StreamAllocator::NeedSpiltNewStream(int64_t stream_node_num, int64_t max_node_num_one_stream, const OpDescPtr &op_desc, bool is_stream_first_node) const { if (is_stream_first_node) { GELOGD("First node of stream does not need to split new stream"); return false; } const set label_op_types({LABELSET, LABELGOTO, LABELGOTOEX, LABELSWITCH, LABELSWITCHBYINDEX}); bool is_first_active_node = false; (void)AttrUtils::GetBool(op_desc, ATTR_NAME_SUBGRAPH_FIRST_ACTIVE, is_first_active_node); return (stream_node_num > max_node_num_one_stream && op_desc->GetSubgraphInstanceNames().empty() && !is_first_active_node && label_op_types.count(op_desc->GetType()) == 0); } Status StreamAllocator::UpdateActiveStreams(const vector> &split_streams) { UpdateLabelStreams(split_streams); for (auto &node : whole_graph_->GetNodes(whole_graph_->GetGraphUnknownFlag())) { if ((node->GetType() == STREAMSWITCH) || (node->GetType() == STREAMSWITCHN)) { if (UpdateActiveStreamsForSwitchNode(node) != SUCCESS) { GELOGE(FAILED, "Update active streams for switch node: %s failed.", node->GetName().c_str()); return FAILED; } } else { if (UpdateActiveStreamsForActiveNode(split_streams, node) != SUCCESS) { GELOGE(FAILED, "Update active streams for active node: %s failed.", node->GetName().c_str()); return FAILED; } } } Status status = UpdateActiveStreamsForSubgraphs(); if (status != SUCCESS) { GELOGE(status, "Update active streams for subgraphs failed!"); return status; } status = SetActiveStreamsForLoop(); if (status != SUCCESS) { GELOGE(status, "SetActiveStreamsForLoop failed!"); return status; } return SUCCESS; } void StreamAllocator::UpdateLabelStreams(const vector> &split_streams) { for (size_t i = 0; i < split_streams.size(); i++) { auto &streams = split_streams[i]; if (streams.empty()) { continue; } if (specific_activated_streams_.count(static_cast(i)) > 0) { specific_activated_streams_.insert(streams.begin(), streams.end()); } for (auto &labeled_stream : labeled_streams_) { if (labeled_stream.second.count(static_cast(i)) > 0) { labeled_stream.second.insert(streams.begin(), streams.end()); break; } } } } Status StreamAllocator::UpdateActiveStreamsForSwitchNode(NodePtr &switch_node) { vector active_nodes; if (InsertActiveNodesAfterSwitch(switch_node, active_nodes) != SUCCESS) { GELOGE(FAILED, "Insert active nodes after node %s failed.", switch_node->GetName().c_str()); return FAILED; } if (active_nodes.empty()) { return SUCCESS; } vector stream_ids; for (auto &active_node : active_nodes) { GE_CHECK_NOTNULL(active_node->GetOpDesc()); active_node->GetOpDesc()->SetStreamId(stream_num_); stream_ids.emplace_back(stream_num_); specific_activated_streams_.emplace(stream_num_); stream_num_++; } auto op_desc = switch_node->GetOpDesc(); GE_CHECK_NOTNULL(op_desc); if (!AttrUtils::SetListInt(op_desc, ATTR_NAME_ACTIVE_STREAM_LIST, stream_ids)) { REPORT_INNER_ERROR("E19999", "Set Attr:%s fail for op:%s(%s)", ATTR_NAME_ACTIVE_STREAM_LIST.c_str(), op_desc->GetName().c_str(), op_desc->GetType().c_str()); GELOGE(FAILED, "SetListInt failed."); return FAILED; } return SUCCESS; } Status StreamAllocator::InsertActiveNodesAfterSwitch(NodePtr &switch_node, vector &active_nodes) { GE_CHECK_NOTNULL(switch_node); OpDescPtr switch_desc = switch_node->GetOpDesc(); GE_CHECK_NOTNULL(switch_desc); vector ori_active_label_list; if (!AttrUtils::GetListStr(switch_desc, ATTR_NAME_ACTIVE_LABEL_LIST, ori_active_label_list) || ori_active_label_list.empty()) { REPORT_INNER_ERROR("E19999", "Get Attr:%s fail for op:%s(%s)", ATTR_NAME_ACTIVE_LABEL_LIST.c_str(), switch_node->GetName().c_str(), switch_node->GetType().c_str()); GELOGE(INTERNAL_ERROR, "Get active label list of switch %s failed.", switch_node->GetName().c_str()); return INTERNAL_ERROR; } vector active_label_list; vector added_active_nodes; if (AddActiveNodes(switch_node, ori_active_label_list, active_label_list, added_active_nodes) != SUCCESS) { GELOGE(FAILED, "Add active nodes after node %s failed.", switch_node->GetName().c_str()); return FAILED; } if (SetActiveLabelList(switch_node, active_label_list) != SUCCESS) { GELOGE(FAILED, "set active label list failed"); return FAILED; } if (added_active_nodes.empty()) { return SUCCESS; } for (auto &active_node : added_active_nodes) { GE_CHECK_NOTNULL(switch_node->GetOutControlAnchor()); if (switch_node->GetOutControlAnchor()->LinkTo(active_node->GetInControlAnchor()) != GRAPH_SUCCESS) { REPORT_CALL_ERROR("E19999", "Link from %s to %s failed", switch_node->GetName().c_str(), active_node->GetName().c_str()); GELOGE(FAILED, "Link %s to %s failed.", switch_node->GetName().c_str(), active_node->GetName().c_str()); return FAILED; } active_nodes.emplace_back(active_node); } return SUCCESS; } Status StreamAllocator::UpdateActiveStreamsForActiveNode(const vector> &split_streams, NodePtr &node) { GE_CHECK_NOTNULL(node->GetOpDesc()); vector active_streams; if (AttrUtils::GetListInt(node->GetOpDesc(), ATTR_NAME_ACTIVE_STREAM_LIST, active_streams)) { vector new_active_streams = active_streams; for (uint32_t logical_stream : active_streams) { if (static_cast(logical_stream) >= split_streams.size()) { REPORT_INNER_ERROR("E19999", "Check logical stream:%u is out of range:%zu", logical_stream, split_streams.size()); GELOGE(FAILED, "logical stream is out of range."); return FAILED; } const set &new_split_streams = split_streams[logical_stream]; for (int64_t split_stream : new_split_streams) { for (const auto &node_stream : node_split_stream_map_) { if (split_stream == node_stream.second) { if (node_stream.first->GetOwnerComputeGraph() == node->GetOwnerComputeGraph()) { new_active_streams.emplace_back(static_cast(split_stream)); GELOGI("Add stream %ld to active_stream_list of node %s of graph %s", split_stream, node->GetName().c_str(), node->GetOwnerComputeGraph()->GetName().c_str()); } break; } } } } if (!AttrUtils::SetListInt(node->GetOpDesc(), ATTR_NAME_ACTIVE_STREAM_LIST, new_active_streams)) { REPORT_INNER_ERROR("E19999", "Set Attr:%s fail for op:%s(%s)", ATTR_NAME_ACTIVE_STREAM_LIST.c_str(), node->GetName().c_str(), node->GetType().c_str()); GELOGE(FAILED, "Set active streams for node %s failed.", node->GetName().c_str()); return FAILED; } } return SUCCESS; } Status StreamAllocator::UpdateActiveStreamsForSubgraphs() const { // Update active stream list for active nodes for (auto &node_stream_pair : node_split_stream_map_) { auto node = node_stream_pair.first; auto subgraph = node->GetOwnerComputeGraph(); if (subgraph->GetParentNode() == nullptr) { continue; } // Skip streams with label GE_CHECK_NOTNULL(node->GetOpDesc()); string stream_label; if (AttrUtils::GetStr(node->GetOpDesc(), ATTR_NAME_STREAM_LABEL, stream_label) && !stream_label.empty()) { continue; } auto it = subgraph_first_active_node_map_.find(subgraph); if (it == subgraph_first_active_node_map_.end()) { continue; } const auto &active_node = it->second; GE_CHECK_NOTNULL(active_node); auto active_op = active_node->GetOpDesc(); GE_CHECK_NOTNULL(active_op); vector active_streams; (void)AttrUtils::GetListInt(active_op, ATTR_NAME_ACTIVE_STREAM_LIST, active_streams); set new_active_streams(active_streams.begin(), active_streams.end()); // specific_activated_streams_ has already contained new split activated stream int64_t new_split_stream = node_stream_pair.second; if (IsActivated(new_split_stream)) { continue; } new_active_streams.emplace(static_cast(new_split_stream)); active_streams.assign(new_active_streams.begin(), new_active_streams.end()); if (!AttrUtils::SetListInt(active_op, ATTR_NAME_ACTIVE_STREAM_LIST, active_streams)) { REPORT_INNER_ERROR("E19999", "Set Attr:%s fail for op:%s(%s)", ATTR_NAME_ACTIVE_STREAM_LIST.c_str(), active_op->GetName().c_str(), active_op->GetType().c_str()); GELOGE(FAILED, "Set active streams for node %s failed.", active_node->GetName().c_str()); return FAILED; } } return SUCCESS; } bool StreamAllocator::IsActivated(int64_t stream_id) const { for (const auto &node : whole_graph_->GetNodes(whole_graph_->GetGraphUnknownFlag())) { auto op_desc = node->GetOpDesc(); vector active_streams; if (op_desc == nullptr || !AttrUtils::GetListInt(op_desc, ATTR_NAME_ACTIVE_STREAM_LIST, active_streams)) { continue; } if (std::find(active_streams.begin(), active_streams.end(), stream_id) != active_streams.end()) { return true; } } return false; } // Iteraotor loop : // StreamSwitch -> StreamActive // FpBp loop: // StreamSwitch -> AssignAdd -> StreamActive NodePtr FindSwitchNodeBeforeLoopActiveNode(const NodePtr &active_node) { for (auto pre_node : active_node->GetInControlNodes()) { if (pre_node->GetType() == STREAMSWITCH) { return pre_node; } for (auto pre_pre_node : pre_node->GetInControlNodes()) { if (pre_pre_node->GetType() == STREAMSWITCH) { return pre_pre_node; } } } return nullptr; } Status StreamAllocator::SetActiveStreamsForLoop() { vector loop_active_streams; for (int64_t stream_id = 0; stream_id < stream_num_; stream_id++) { if (specific_activated_streams_.count(stream_id) == 0) { loop_active_streams.emplace_back(static_cast(stream_id)); } } map stream_id_to_last_node; set streams_skip_iterator_event; for (const auto &node : whole_graph_->GetNodes(whole_graph_->GetGraphUnknownFlag())) { int64_t stream_id = node->GetOpDesc()->GetStreamId(); if (find(loop_active_streams.begin(), loop_active_streams.end(), stream_id) != loop_active_streams.end()) { stream_id_to_last_node[stream_id] = node; // last node in stream which has streamswitch or IF may be not execute, it will cause block if add event on them if (node->GetOpDesc()->GetType() == STREAMSWITCH) { streams_skip_iterator_event.insert(stream_id); } } } // Set the stream that needs to be activated for (const auto &node : whole_graph_->GetNodes(whole_graph_->GetGraphUnknownFlag())) { GE_CHECK_NOTNULL(node->GetOpDesc()); bool is_loop_active = false; if (AttrUtils::GetBool(node->GetOpDesc(), ATTR_NAME_IS_LOOP_ACTIVE, is_loop_active) && is_loop_active) { vector activated_label_list; NodePtr pre_switch_node = FindSwitchNodeBeforeLoopActiveNode(node); if (pre_switch_node == nullptr) { REPORT_INNER_ERROR("E19999", "Find switch node before loop active node %s fail", node->GetName().c_str()); GELOGE(FAILED, "find switch node before loop active node %s failed", node->GetName().c_str()); return FAILED; } if (!AttrUtils::GetListStr(node->GetOpDesc(), ATTR_NAME_ACTIVE_LABEL_LIST, activated_label_list) || activated_label_list.empty()) { GE_CHK_BOOL_EXEC(AttrUtils::SetListInt(node->GetOpDesc(), ATTR_NAME_ACTIVE_STREAM_LIST, loop_active_streams), REPORT_INNER_ERROR("E19999", "Set Attr:%s fail for op:%s(%s)", ATTR_NAME_ACTIVE_STREAM_LIST.c_str(), node->GetName().c_str(), node->GetType().c_str()); GELOGE(FAILED, "SetListInt failed."); return FAILED); for (const auto &stream_id : loop_active_streams) { GELOGI("Active stream %u for node: %s.", stream_id, node->GetName().c_str()); } // In switch group optimze case, some data input branch may exec slowly. // when condition input branch judge false and some switch has no false branch, // In this condition, data branch has no synchronize point, // it may cause some stream actived by iterator next step when this stream still alive. // If above situation happen, active message will lose, cause process block in next iteration. // In order to avoid this abnormal happen, // add event between each last node and iterator switch node GELOGI("there are %zu next iterator target streams has streamswitch node.", streams_skip_iterator_event.size()); for (auto iter : stream_id_to_last_node) { if (streams_skip_iterator_event.find(iter.first) != streams_skip_iterator_event.end()) { GELOGI("Skip stream %ld which has streamswitch node when adding event to next iterator active node", iter.first); continue; } if (iter.second->GetOwnerComputeGraph()->GetParentGraph() != nullptr) { GELOGI("Skip stream %ld which is last node in subgraph when adding event to next iterator active node", iter.first); continue; } AddSendEventId(iter.second, event_num_); AddRecvEventId(pre_switch_node, event_num_); event_num_++; } break; } } } return CheckStreamActived(); } Status StreamAllocator::CheckStreamActived() const { for (const auto &node : whole_graph_->GetNodes(whole_graph_->GetGraphUnknownFlag())) { GE_CHECK_NOTNULL(node->GetOpDesc()); vector active_streams; if (AttrUtils::GetListInt(node->GetOpDesc(), ATTR_NAME_ACTIVE_STREAM_LIST, active_streams)) { uint32_t stream_id = static_cast(node->GetOpDesc()->GetStreamId()); auto iter = find(active_streams.begin(), active_streams.end(), stream_id); if (iter != active_streams.end()) { REPORT_INNER_ERROR("E19999", "Node:%s(%s) cannot active its own stream %u, check invalid ", node->GetName().c_str(), node->GetType().c_str(), stream_id); GELOGE(FAILED, "Node %s cannot active its own stream %u.", node->GetName().c_str(), stream_id); return FAILED; } } } return SUCCESS; } Status StreamAllocator::ReuseEvent(bool send_to, const std::unordered_map &name_to_node_map, const std::unordered_map>> &node_to_event_id) { for (const auto &node_event_id : node_to_event_id) { ge::NodePtr curr_node = node_event_id.first; NodePtr send_node = send_to ? curr_node : nullptr; NodePtr recv_node = send_to ? nullptr : curr_node; for (const auto &event_pair : node_event_id.second) { auto peer_node_iter = name_to_node_map.find(event_pair.first); if (peer_node_iter == name_to_node_map.end()) { GELOGE(PARAM_INVALID, "[Get][Node]Name:%s.", event_pair.first.c_str()); REPORT_INNER_ERROR("E19999", "Failed to find node, name:%s.", event_pair.first.c_str()); return PARAM_INVALID; } recv_node = send_to ? peer_node_iter->second : recv_node; send_node = send_to ? send_node : peer_node_iter->second; GE_CHECK_NOTNULL(send_node); GE_CHECK_NOTNULL(recv_node); auto event_id = GetIntersection(node_to_send_events_[send_node], node_to_recv_events_[recv_node]); uint32_t new_event = event_pair.second + event_num_; if (event_id.empty()) { GELOGI("[Check][Optimized]Send:%s, recv:%s.", send_node->GetName().c_str(), recv_node->GetName().c_str()); continue; } else if (event_id.size() != 1) { GELOGW("[Check][Event]More than one event are found between %s and %s, event num:%zu.", send_node->GetName().c_str(), recv_node->GetName().c_str(), event_id.size()); } uint32_t old_event = event_id[0]; auto reuse_event_id = [] (vector &event_list, uint32_t old_event, uint32_t new_event) -> void { event_list.erase(std::remove(event_list.begin(), event_list.end(), old_event), event_list.end()); event_list.push_back(new_event); return; }; reuse_event_id(node_to_send_events_[send_node], old_event, new_event); reuse_event_id(node_to_recv_events_[recv_node], old_event, new_event); GELOGI("[Reuse][Event]Replace event successfully, send node:%s, recv node:%s, old id:%u, new id:%u.", send_node->GetName().c_str(), recv_node->GetName().c_str(), old_event, new_event); } } return ge::SUCCESS; } // Refresh events to reuse events Status StreamAllocator::RefreshEventsWithReuse() { GELOGI("[Refresh][Events]Refresh events with reuse, stream num:%ld, original event num:%u.", stream_num_, event_num_); if (event_num_ <= kEventReuseThreshold) { GELOGI("[Check][ReuseThreshold]Event used num is %u, less than %u, skip reuse.", event_num_, kEventReuseThreshold); return SUCCESS; } std::unordered_map name_to_node_map; std::unordered_map>> node_to_send; std::unordered_map>> node_to_recv; Status ret = ParseAllNodeEventMultiplexing(whole_graph_, name_to_node_map, node_to_send, node_to_recv); if (ret != SUCCESS) { GELOGE(ret, "[Parse][AllNodeEventMultiplexing]Graph:%s.", whole_graph_->GetName().c_str()); REPORT_INNER_ERROR("E19999", "Failed to parse all node event multiplexing, graph:%s.", whole_graph_->GetName().c_str()); return ret; } if (node_to_send.empty() && node_to_recv.empty()) { return SUCCESS; } ret = ReuseEvent(true, name_to_node_map, node_to_send); if (ret != SUCCESS) { GELOGE(ret, "[Reuse][Event]Phase:Send, graph:%s.", whole_graph_->GetName().c_str()); REPORT_INNER_ERROR("E19999", "Failed to reuse event, phase:Send, graph:%s.", whole_graph_->GetName().c_str()); return ret; } ret = ReuseEvent(false, name_to_node_map, node_to_recv); if (ret != SUCCESS) { GELOGE(ret, "[Reuse][Event]Phase:Recv, graph:%s.", whole_graph_->GetName().c_str()); REPORT_INNER_ERROR("E19999", "Failed to reuse event, phase:Recv, graph:%s.", whole_graph_->GetName().c_str()); return ret; } Status status = RefreshContinuousEvents(); if (status != SUCCESS) { GELOGE(status, "[Refresh][ContinuousEvents]Graph:%s.", whole_graph_->GetName().c_str()); REPORT_INNER_ERROR("E19999", "Failed to refresh continuous events, graph:%s.", whole_graph_->GetName().c_str()); return status; } GELOGI("[Refresh][Events]RefreshEventsWithReuse successfully, event num:%u.", event_num_); return SUCCESS; } // Refresh events to continuous events Status StreamAllocator::RefreshContinuousEvents() { // Establish a mapping relationship from old to new event id map old_to_new_events; uint32_t new_event_id = 0; for (const auto &one_pair : node_to_send_events_) { for (const auto &event_id : one_pair.second) { if (old_to_new_events.find(event_id) == old_to_new_events.end()) { old_to_new_events[event_id] = new_event_id; new_event_id++; } } } // Refresh send event id for (auto &one_pair : node_to_send_events_) { vector &send_events = one_pair.second; for (size_t i = 0; i < send_events.size(); i++) { auto find_it = old_to_new_events.find(send_events[i]); if (find_it == old_to_new_events.end()) { REPORT_INNER_ERROR("E19999", "Check invalid send event %u", send_events[i]); GELOGE(FAILED, "RefreshContinuousEvents: invalid send event %u", send_events[i]); return FAILED; } send_events[i] = find_it->second; } } // Refresh recv event id for (auto &one_pair : node_to_recv_events_) { vector &recv_events = one_pair.second; for (size_t i = 0; i < recv_events.size(); i++) { auto find_it = old_to_new_events.find(recv_events[i]); if (find_it == old_to_new_events.end()) { REPORT_INNER_ERROR("E19999", "Check invalid recv event %u", recv_events[i]); GELOGE(FAILED, "RefreshContinuousEvents: invalid recv event %u", recv_events[i]); return FAILED; } recv_events[i] = find_it->second; } } event_num_ = static_cast(old_to_new_events.size()); return SUCCESS; } // Insert the real send/recv node in the graph Status StreamAllocator::InsertSyncEventNodes() { unordered_map sync_event_name; for (const auto &node : whole_graph_->GetNodes(whole_graph_->GetGraphUnknownFlag())) { // Add the node corresponding to the recv event vector recv_event_id_list; GetRecvEventIdList(node, recv_event_id_list); GE_CHECK_NOTNULL(node->GetOpDesc()); GE_CHECK_NOTNULL(node->GetInControlAnchor()); GE_CHECK_NOTNULL(node->GetOutControlAnchor()); for (auto &event_id : recv_event_id_list) { string recv_node_name = whole_graph_->GetName() + "_Recv_" + to_string(event_id); auto iter = sync_event_name.find(recv_node_name); if (iter == sync_event_name.end()) { sync_event_name[recv_node_name] = 1; } else { recv_node_name = recv_node_name + "_Reuse_" + to_string(iter->second); ++(iter->second); } OpDescPtr op_desc_ptr = MakeShared(recv_node_name, RECV); GE_CHECK_NOTNULL(op_desc_ptr); int64_t temp_stream_id = node->GetOpDesc()->GetStreamId(); op_desc_ptr->SetStreamId(temp_stream_id); GE_CHK_BOOL_EXEC(AttrUtils::SetInt(op_desc_ptr, RECV_ATTR_EVENT_ID, event_id), REPORT_INNER_ERROR("E19999", "Set Attr:%s for op:%s(%s) failed, event_id:%u,", RECV_ATTR_EVENT_ID.c_str(), node->GetName().c_str(), node->GetType().c_str(), event_id); GELOGE(FAILED, "SetInt failed."); return FAILED); (void)AttrUtils::SetListStr(op_desc_ptr, ATTR_NAME_DATA_DUMP_ORIGIN_OP_NAMES, std::move(std::vector())); NodePtr recv_node = node->GetOwnerComputeGraph()->AddNode(op_desc_ptr); GE_CHECK_NOTNULL(recv_node); GE_CHECK_NOTNULL(recv_node->GetOutControlAnchor()); Status status = GraphUtils::AddEdge(recv_node->GetOutControlAnchor(), node->GetInControlAnchor()); if (status != SUCCESS) { REPORT_INNER_ERROR("E19999", "Add edge from node %s to node %s failed", recv_node->GetName().c_str(), node->GetName().c_str()); GELOGE(status, "Add edge for node %s and node %s failed.", recv_node->GetName().c_str(), node->GetName().c_str()); return status; } GELOGI("Insert recv event %u before node: %s.", event_id, node->GetName().c_str()); } // Add the node corresponding to the send event vector send_event_id_list; GetSendEventIdList(node, send_event_id_list); for (auto &event_id : send_event_id_list) { string send_node_name = whole_graph_->GetName() + "_Send_" + to_string(event_id); auto iter = sync_event_name.find(send_node_name); if (iter == sync_event_name.end()) { sync_event_name[send_node_name] = 1; } else { send_node_name = send_node_name + "_Reuse_" + to_string(iter->second); ++(iter->second); } OpDescPtr op_desc_ptr = MakeShared(send_node_name, SEND); GE_CHECK_NOTNULL(op_desc_ptr); int64_t temp_stream_id = node->GetOpDesc()->GetStreamId(); op_desc_ptr->SetStreamId(temp_stream_id); GE_CHK_BOOL_EXEC(AttrUtils::SetInt(op_desc_ptr, SEND_ATTR_EVENT_ID, event_id), GELOGE(FAILED, "SetInt failed."); return FAILED); (void)AttrUtils::SetListStr(op_desc_ptr, ATTR_NAME_DATA_DUMP_ORIGIN_OP_NAMES, std::move(std::vector())); NodePtr send_node = node->GetOwnerComputeGraph()->AddNode(op_desc_ptr); GE_CHECK_NOTNULL(send_node); GE_CHECK_NOTNULL(send_node->GetInControlAnchor()); Status status = GraphUtils::AddEdge(node->GetOutControlAnchor(), send_node->GetInControlAnchor()); if (status != SUCCESS) { REPORT_INNER_ERROR("E19999", "Add edge from node %s to node %s failed", node->GetName().c_str(), send_node->GetName().c_str()); GELOGE(status, "Add edge for node %s and node %s failed.", node->GetName().c_str(), send_node->GetName().c_str()); return status; } GELOGI("Insert send event %u after node: %s.", event_id, node->GetName().c_str()); } } Status status = whole_graph_->InsertGraphEvents(); if (status != SUCCESS) { REPORT_CALL_ERROR("E19999", "Insert Graph Events fail, graph:%s,", whole_graph_->GetName().c_str()); GELOGE(status, "Graph ReorderEventNodes failed"); return status; } return SUCCESS; } void StreamAllocator::DumpEvents() { map> after_refresh_stream_nodes; for (const auto &node : whole_graph_->GetNodes(whole_graph_->GetGraphUnknownFlag())) { GE_IF_BOOL_EXEC(node->GetOpDesc() == nullptr, continue); int64_t stream_id = node->GetOpDesc()->GetStreamId(); after_refresh_stream_nodes[stream_id].emplace_back(node); } for (const auto &one_pair : after_refresh_stream_nodes) { int64_t stream_id = one_pair.first; GELOGD("After RefreshRealStream: stream %ld.", stream_id); for (const auto &node : one_pair.second) { if (node == nullptr || node->GetOpDesc() == nullptr) { continue; } string send_event_str; for (const auto &send_event_id : node_to_send_events_[node]) { send_event_str += " " + to_string(send_event_id); } if (!send_event_str.empty()) { GELOGI("node: %s, id: %ld, stream id :%ld, send events: %s.", node->GetName().c_str(), node->GetOpDesc()->GetId(), node->GetOpDesc()->GetStreamId(), send_event_str.c_str()); } string recv_event_str; for (const auto &recv_event_id : node_to_recv_events_[node]) { recv_event_str += " " + to_string(recv_event_id); } if (!recv_event_str.empty()) { GELOGI("node: %s, id: %ld, stream id :%ld, recv events: %s.", node->GetName().c_str(), node->GetOpDesc()->GetId(), node->GetOpDesc()->GetStreamId(), recv_event_str.c_str()); } } } } Status StreamAllocator::GetMaxStreamAndTask(bool huge_stream, uint32_t &max_stream_count, uint32_t &max_task_count) { uint32_t stream_type = RT_NORMAL_STREAM; if (huge_stream) { stream_type = RT_HUGE_STREAM; } rtError_t ret = rtGetMaxStreamAndTask(stream_type, &max_stream_count, &max_task_count); if (ret != RT_ERROR_NONE) { REPORT_CALL_ERROR("E19999", "call rtGetMaxStreamAndTask fail, ret:%d, stream_type:%u,", static_cast(ret), stream_type); GELOGE(FAILED, "Get max stream and task count by rts failed."); return FAILED; } GELOGD("Allowed max stream count: %u, max task count per stream: %u.", max_stream_count, max_task_count); return SUCCESS; } int64_t StreamAllocator::GetMaxNodeNumPerStream(const NodePtr &node, uint32_t max_task_count) { int64_t max_node_num_one_stream = static_cast(max_task_count); string op_type = node->GetType(); if (IsHcclOp(op_type)) { max_node_num_one_stream /= kTaskNumPerHcclNode; } else { max_node_num_one_stream /= kTaskNumPerNormalNode; } if (max_node_num_one_stream == 0) { max_node_num_one_stream = 1; } return max_node_num_one_stream; } void StreamAllocator::AddNodeNum(const NodePtr &node, int64_t &node_num) { node_num++; vector events; GetSendEventIdList(node, events); node_num += static_cast(events.size()); GetRecvEventIdList(node, events); node_num += static_cast(events.size()); } // Insert send event id on a node void StreamAllocator::AddSendEventId(const NodePtr &node, uint32_t event_id) { node_to_send_events_[node].emplace_back(event_id); } // Insert recv event id on a node void StreamAllocator::AddRecvEventId(const NodePtr &node, uint32_t event_id) { node_to_recv_events_[node].emplace_back(event_id); } // Remove send event id from a node void StreamAllocator::RmvSendEventId(const NodePtr &node, uint32_t event_id) { auto find_it = node_to_send_events_.find(node); if (find_it == node_to_send_events_.end()) { return; } vector &send_events = find_it->second; for (auto it = send_events.begin(); it != send_events.end(); ++it) { if (*it == event_id) { send_events.erase(it); return; } } } // Remove recv event id from a node void StreamAllocator::RmvRecvEventId(const NodePtr &node, uint32_t event_id) { auto find_it = node_to_recv_events_.find(node); if (find_it == node_to_recv_events_.end()) { return; } vector &recv_events = find_it->second; for (auto it = recv_events.begin(); it != recv_events.end(); ++it) { if (*it == event_id) { recv_events.erase(it); return; } } } // Get send event id list from a node void StreamAllocator::GetSendEventIdList(const NodePtr &node, vector &send_list) const { send_list.clear(); auto find_it = node_to_send_events_.find(node); if (find_it != node_to_send_events_.end()) { send_list = find_it->second; } } // Get recv event id list from a node void StreamAllocator::GetRecvEventIdList(const NodePtr &node, vector &recv_list) const { recv_list.clear(); auto find_it = node_to_recv_events_.find(node); if (find_it != node_to_recv_events_.end()) { recv_list = find_it->second; } } // Get a specific send node according to the recv event NodePtr StreamAllocator::GetNodeFromSendEventId(uint32_t send_event_id) const { for (const auto &one_pair : node_to_send_events_) { const vector &events = one_pair.second; for (const auto &event_id : events) { if (event_id == send_event_id) { return one_pair.first; } } } return nullptr; } // Get a specific recv node according to the recv event NodePtr StreamAllocator::GetNodeFromRecvEventId(uint32_t recv_event_id) const { for (const auto &one_pair : node_to_recv_events_) { const vector &events = one_pair.second; for (const auto &event_id : events) { if (event_id == recv_event_id) { return one_pair.first; } } } return nullptr; } Status StreamAllocator::AddEventId(const NodePtr &pre_node, const NodePtr ¬_cur, const NodePtr &cur_node, bool not_use_cur) { GELOGI("Add send event %u for node %s", event_num_, pre_node->GetName().c_str()); AddSendEventId(pre_node, event_num_); if (not_use_cur) { GE_CHECK_NOTNULL(not_cur); GELOGI("Add recv event %u for node %s", event_num_, not_cur->GetName().c_str()); AddRecvEventId(not_cur, event_num_); } else { GELOGI("Add recv event %u for node %s", event_num_, cur_node->GetName().c_str()); AddRecvEventId(cur_node, event_num_); } ++event_num_; return SUCCESS; } Status StreamAllocator::AddActiveNodes(NodePtr &switch_node, const vector &ori_active_label_list, vector &active_label_list, vector &added_active_nodes) { size_t label_num = ori_active_label_list.size(); for (size_t i = 0; i < label_num; i++) { const string &active_label = ori_active_label_list[i]; if (labeled_streams_.find(active_label) == labeled_streams_.end()) { REPORT_INNER_ERROR("E19999", "can not find stream label:%s", active_label.c_str()); GELOGE(FAILED, "can not find stream label %s", active_label.c_str()); return FAILED; } if (labeled_streams_[active_label].size() <= 1) { active_label_list.emplace_back(active_label); continue; } string name = switch_node->GetName() + "_" + STREAMACTIVE + "_" + std::to_string(i); GELOGI("Create StreamActive op %s after node %s.", name.c_str(), switch_node->GetName().c_str()); OpDescPtr active_op_desc = MakeShared(name, STREAMACTIVE); GE_CHECK_NOTNULL(active_op_desc); NodePtr active_node = whole_graph_->AddNode(active_op_desc); GE_CHECK_NOTNULL(active_node); for (NodePtr &node : switch_node->GetOutControlNodes()) { OpDescPtr op_desc = node->GetOpDesc(); GE_CHECK_NOTNULL(op_desc); string stream_label; // If GetStr failed, stream_label is empty. (void)AttrUtils::GetStr(op_desc, ATTR_NAME_STREAM_LABEL, stream_label); if (stream_label != active_label) { continue; } GE_CHECK_NOTNULL(switch_node->GetOutControlAnchor()); if (switch_node->GetOutControlAnchor()->Unlink(node->GetInControlAnchor()) != GRAPH_SUCCESS) { REPORT_CALL_ERROR("E19999", "Unlink %s to %s failed", switch_node->GetName().c_str(), node->GetName().c_str()); GELOGE(FAILED, "Unlink %s to %s failed.", switch_node->GetName().c_str(), node->GetName().c_str()); return FAILED; } GE_CHECK_NOTNULL(active_node->GetOutControlAnchor()); if (active_node->GetOutControlAnchor()->LinkTo(node->GetInControlAnchor()) != GRAPH_SUCCESS) { REPORT_CALL_ERROR("E19999", "Link %s to %s failed", active_node->GetName().c_str(), node->GetName().c_str()); GELOGE(FAILED, "Link %s to %s failed.", active_node->GetName().c_str(), node->GetName().c_str()); return FAILED; } } if (SetSwitchBranchNodeLabel(active_node, name) != SUCCESS) { GELOGE(FAILED, "Set switch branch node label failed."); return FAILED; } if (SetStreamLabel(active_node, name) != SUCCESS) { GELOGE(FAILED, "Set stream label failed."); return FAILED; } if (SetActiveLabelList(active_node, {active_label}) != SUCCESS) { GELOGE(FAILED, "Set active label list failed."); return FAILED; } if (SetActiveStreamList(active_node, active_label) != SUCCESS) { GELOGE(FAILED, "Set active stream list failed."); return FAILED; } added_active_nodes.emplace_back(active_node); active_label_list.emplace_back(name); } return SUCCESS; } Status StreamAllocator::SetActiveStreamList(NodePtr &active_node, const string &active_label) { if (labeled_streams_.find(active_label) == labeled_streams_.end()) { REPORT_INNER_ERROR("E19999", "Can not find stream label:%s", active_label.c_str()); GELOGE(FAILED, "Can not find stream label %s.", active_label.c_str()); return FAILED; } set &streams = labeled_streams_[active_label]; vector active_streams(streams.begin(), streams.end()); if (!AttrUtils::SetListInt(active_node->GetOpDesc(), ATTR_NAME_ACTIVE_STREAM_LIST, active_streams)) { REPORT_INNER_ERROR("E19999", "Set Attr:%s fail for op:%s(%s)", ATTR_NAME_ACTIVE_STREAM_LIST.c_str(), active_node->GetName().c_str(), active_node->GetType().c_str()); GELOGE(FAILED, "SetListInt of %s failed.", ATTR_NAME_ACTIVE_STREAM_LIST.c_str()); return FAILED; } return SUCCESS; } } // namespace ge