diff --git a/ge/graph/execute/graph_execute.cc b/ge/graph/execute/graph_execute.cc index 03abf91f..c97a4fbc 100755 --- a/ge/graph/execute/graph_execute.cc +++ b/ge/graph/execute/graph_execute.cc @@ -434,27 +434,10 @@ Status GraphExecutor::ExecuteGraphWithStream(GraphId graph_id, GE_CHECK_NOTNULL_EXEC(ge_root_model, return FAILED); auto model_id = ge_root_model->GetModelId(); - InputData input_data; - input_data.index = 0; - input_data.model_id = model_id; - std::vector input_desc; - auto ret = GetExecuteData(input_tensor, input_data.blobs, input_desc); - if (ret != SUCCESS) { - return ret; - } - OutputData output_data; - output_data.index = 0; - output_data.model_id = model_id; - std::vector output_desc; - ret = GetExecuteData(output_tensor, output_data.blobs, output_desc); - if (ret != SUCCESS) { - return ret; - } - auto async_mode = true; auto model_manager = ge::ModelManager::GetInstance(); GE_CHECK_NOTNULL(model_manager); - ret = model_manager->ExecuteModel(model_id, stream, async_mode, input_data, input_desc, output_data, output_desc); + auto ret = model_manager->ExecuteModel(model_id, stream, async_mode, input_tensor, output_tensor); if (ret != SUCCESS) { return ret; } diff --git a/ge/graph/load/model_manager/davinci_model.cc b/ge/graph/load/model_manager/davinci_model.cc index 495ec28e..ec814eaf 100755 --- a/ge/graph/load/model_manager/davinci_model.cc +++ b/ge/graph/load/model_manager/davinci_model.cc @@ -3530,6 +3530,31 @@ Status DavinciModel::CopyModelData(const InputData &input_data, OutputData &outp return SUCCESS; } +void DavinciModel::BuildZeroCopyTasksLookupTable() { + std::lock_guard lk(lookup_table_build_lock_); + if (lookup_table_built_) { + return; + } + + const auto default_label_hash = std::hash{}(kDefaultBatchLable); + for (auto &task : zero_copy_tasks_) { + auto label_hash = std::hash{}(task.GetBatchLabel()); + auto addr2offsets = task.GetTaskArgsOffset(); + + label_hash2tasks_[label_hash].insert(&task); + if (label_hash == default_label_hash) { + for (auto &addr2offset : addr2offsets) { + addr2default_label_tasks_[addr2offset.first].insert(&task); + } + } else { + for (auto &addr2offset : addr2offsets) { + addr2specific_label_tasks_[addr2offset.first].insert(&task); + } + } + } + lookup_table_built_ = true; +} + /// /// @ingroup ge /// @brief Copy Data addr to model for direct use. @@ -3551,6 +3576,8 @@ Status DavinciModel::UpdateIoTaskArgs(const std::map & return ACL_ERROR_GE_PARAM_INVALID; } + BuildZeroCopyTasksLookupTable(); + for (const auto &data : data_info) { if (data.first >= blobs.size()) { // check data index. REPORT_INNER_ERROR("E19999", "is_input:%d, data index:%u from model >= blobs.size:%zu from user, mode_id:%u" @@ -3592,21 +3619,34 @@ Status DavinciModel::UpdateIoTaskArgs(const std::map & continue; } + const static auto kDefaultLabelHash = std::hash{}(kDefaultBatchLable); + auto batch_label_hash = std::hash{}(batch_label); + std::unordered_set same_batch_label_tasks; + if (batch_label_hash != kDefaultLabelHash) { + auto iter = label_hash2tasks_.find(batch_label_hash); + if (iter != label_hash2tasks_.end()) { + same_batch_label_tasks = iter->second; + } + } + for (size_t count = 0; count < data.second.GetDataCount(); ++count) { - void *addr = data.second.GetDataInfo().at(count).second; + auto addr = reinterpret_cast(data.second.GetDataInfo().at(count).second); void *buffer_addr = reinterpret_cast(reinterpret_cast(buffer.data) + data.second.GetRelativeOffset().at(count)); - GELOGI("[ZCPY] Copy %s blobs_index %u, virtual_addr: %p, size: %ld, user_data_addr: %p, batch_label: %s", + GELOGI("[ZCPY] Copy %s blobs_index %u, virtual_addr: 0x%lx, size: %ld, user_data_addr: %p, batch_label: %s", is_input ? "input" : "output", data.first, addr, data.second.GetDataInfo().at(count).first, buffer_addr, batch_label.c_str()); // For input data, just copy for rts task. - for (auto &task : zero_copy_tasks_) { - bool not_same_batch = (task.GetBatchLabel() != kDefaultBatchLable && task.GetBatchLabel() != batch_label); - if (not_same_batch) { - continue; + for (auto &task : addr2default_label_tasks_[addr]) { // always update default label tasks + (void)task->UpdateTaskParam(addr, buffer_addr); + } + + if (batch_label_hash != kDefaultLabelHash) { + for (auto &task : addr2specific_label_tasks_[addr]) { + if (same_batch_label_tasks.count(task) > 0) { + (void)task->UpdateTaskParam(addr, buffer_addr); + } } - uintptr_t addr_val = reinterpret_cast(addr); - (void)task.UpdateTaskParam(addr_val, buffer_addr); } } } diff --git a/ge/graph/load/model_manager/davinci_model.h b/ge/graph/load/model_manager/davinci_model.h index 76b0beef..dc9fba85 100755 --- a/ge/graph/load/model_manager/davinci_model.h +++ b/ge/graph/load/model_manager/davinci_model.h @@ -917,6 +917,7 @@ class DavinciModel { Status GetGearAndRealOutSizeInfo(const ComputeGraphPtr &graph, const NodePtr &node); Status GetRealOutputSizeOfCase(const ComputeGraphPtr &graph, size_t input_index, const NodePtr &case_node); Status GetGearAndRealOutShapeInfo(const ComputeGraphPtr &graph, const NodePtr &node); + void BuildZeroCopyTasksLookupTable(); bool is_weight_mem_has_inited_; bool is_feature_map_mem_has_inited_; @@ -1112,6 +1113,13 @@ class DavinciModel { // op name to attrs mapping std::map>> op_name_to_attrs_; + // fields for build fast search hash table for zero copy tasks + std::mutex lookup_table_build_lock_; + bool lookup_table_built_{false}; + std::unordered_map> label_hash2tasks_; + std::unordered_map> addr2specific_label_tasks_; + std::unordered_map> addr2default_label_tasks_; + std::map stream_2_event_; }; } // namespace ge diff --git a/ge/graph/load/model_manager/model_manager.cc b/ge/graph/load/model_manager/model_manager.cc index d0d88e66..3e45300a 100755 --- a/ge/graph/load/model_manager/model_manager.cc +++ b/ge/graph/load/model_manager/model_manager.cc @@ -1332,6 +1332,78 @@ Status ModelManager::ExecuteModel(uint32_t model_id, rtStream_t stream, bool asy return status; } +namespace { +void GetGeTensorBlobs(const std::vector &tensors, std::vector &blobs) { + blobs.resize(tensors.size()); + for (size_t i = 0; i < tensors.size(); i++) { + auto &tensor = tensors[i]; + auto &buf = blobs[i]; + buf.data = const_cast(tensor.GetData().data()); + buf.length = tensor.GetData().size(); + buf.isDataSupportMemShare = false; + } +} + +void GetGeTensorDescs(const std::vector &tensors, std::vector &descs) { + descs.reserve(tensors.size()); + for (auto &tensor : tensors) { + descs.emplace_back(std::move(tensor.GetTensorDesc())); + } +} +} + +ge::Status ModelManager::ExecuteModel(uint32_t model_id, rtStream_t stream, bool async_mode, + const std::vector &input_tensor, std::vector &output_tensor) { + InputData input_data; + input_data.index = 0; + input_data.model_id = model_id; + + OutputData output_data; + output_data.index = 0; + output_data.model_id = model_id; + + GetGeTensorBlobs(input_tensor, input_data.blobs); + GetGeTensorBlobs(output_tensor, output_data.blobs); + + std::shared_ptr hybrid_davinci_model = GetHybridModel(model_id); + if (hybrid_davinci_model != nullptr) { + std::vector input_desc; + std::vector output_desc; + GetGeTensorDescs(input_tensor, input_desc); + GetGeTensorDescs(output_tensor, output_desc); + + Status status = hybrid_davinci_model->Execute(input_data.blobs, input_desc, output_data.blobs, output_desc, stream); + if (status == SUCCESS) { + GELOGI("Execute model %u success.", model_id); + } + return status; + } + + std::shared_ptr davinci_model = GetModel(model_id); + GE_CHK_BOOL_RET_STATUS(davinci_model != nullptr, ACL_ERROR_GE_EXEC_MODEL_ID_INVALID, + "[Get][Model] Invalid model id %u, check whether model has been loaded or not.", model_id); + + if (davinci_model->NeedDestroyAicpuKernel()) { + GELOGI("Start to destroy specified aicpu kernel."); + // Zero copy is enabled by default, no need to judge. + uint64_t session_id_davinci = davinci_model->GetSessionId(); + uint32_t model_id_davinci = davinci_model->GetModelId(); + uint32_t sub_model_id = davinci_model->SubModelId(); + Status status = DestroyAicpuKernel(session_id_davinci, model_id_davinci, sub_model_id); + if (status != SUCCESS) { + GELOGW("Destroy specified aicpu kernel failed, session id is %lu, model id is %u.", session_id_davinci, + model_id_davinci); + } + } + + Status status = davinci_model->NnExecute(stream, async_mode, input_data, output_data); + if (status == SUCCESS) { + GELOGD("Execute model %u success.", model_id); + } + + return status; +} + Status ModelManager::CreateAicpuSession(uint64_t session_id) { std::lock_guard lock(map_mutex_); auto it = sess_ids_.find(session_id); diff --git a/ge/graph/load/model_manager/model_manager.h b/ge/graph/load/model_manager/model_manager.h index 6389d6db..b160ba3b 100755 --- a/ge/graph/load/model_manager/model_manager.h +++ b/ge/graph/load/model_manager/model_manager.h @@ -157,6 +157,9 @@ class FMK_FUNC_HOST_VISIBILITY FMK_FUNC_DEV_VISIBILITY ModelManager { const std::vector &input_desc, OutputData &output_data, std::vector &output_desc); + ge::Status ExecuteModel(uint32_t model_id, rtStream_t stream, bool async_mode, const std::vector &inputs, + std::vector &outputs); + ge::Status SyncExecuteModel(uint32_t model_id, const std::vector &inputs, std::vector &outputs); /// diff --git a/ge/graph/load/model_manager/zero_copy_task.cc b/ge/graph/load/model_manager/zero_copy_task.cc index 85be6d7b..2e4c212c 100755 --- a/ge/graph/load/model_manager/zero_copy_task.cc +++ b/ge/graph/load/model_manager/zero_copy_task.cc @@ -54,6 +54,10 @@ Status ZeroCopyTask::SetTaskArgsOffset(uintptr_t addr, size_t offset) { return SUCCESS; } +const std::map>& ZeroCopyTask::GetTaskArgsOffset() const { + return task_addr_offset_; +} + /** * @ingroup ge * @brief Save orignal data of task args. @@ -80,17 +84,16 @@ void ZeroCopyTask::SetOriginalArgs(const void *info, size_t size) { Status ZeroCopyTask::UpdateTaskParam(uintptr_t addr, void *buffer_addr) { auto iter = task_addr_offset_.find(addr); if (iter != task_addr_offset_.end()) { - auto &cur_pair = *iter; + auto dst_addr = reinterpret_cast(static_cast(buffer_addr)); uint8_t *args_info = args_info_.data(); - for (auto offset : cur_pair.second) { - auto dst_addr = static_cast(buffer_addr); - GELOGI("[ZCPY] %s update task, args_addr: %p, size: %zu, offset: %zu, virtual_addr: 0x%lx, user_data_addr: %p", - name_.c_str(), args_addr_, args_size_, offset, addr, buffer_addr); - *reinterpret_cast(args_info + offset)= reinterpret_cast(dst_addr); - is_updated_ = true; + for (auto offset : iter->second) { + auto ¤t_addr = *reinterpret_cast(args_info + offset); + if (current_addr != dst_addr) { + current_addr = dst_addr; + is_updated_ = true; + } } } - return SUCCESS; } diff --git a/ge/graph/load/model_manager/zero_copy_task.h b/ge/graph/load/model_manager/zero_copy_task.h index efabc814..6420c91c 100644 --- a/ge/graph/load/model_manager/zero_copy_task.h +++ b/ge/graph/load/model_manager/zero_copy_task.h @@ -46,6 +46,8 @@ class ZeroCopyTask { */ ge::Status SetTaskArgsOffset(uintptr_t addr, size_t offset); + const std::map>& GetTaskArgsOffset() const; + /** * @ingroup ge * @brief Is need zero copy. diff --git a/ge/session/inner_session.cc b/ge/session/inner_session.cc index b9c44ef1..8a402c77 100755 --- a/ge/session/inner_session.cc +++ b/ge/session/inner_session.cc @@ -341,12 +341,14 @@ Status InnerSession::RunGraphWithStreamAsync(uint32_t graph_id, rtStream_t strea } UpdateThreadContext(graph_id); vector ge_inputs; + ge_inputs.reserve(inputs.size()); for (auto &item : inputs) { - ge_inputs.emplace_back(TensorAdapter::AsGeTensor(item)); + ge_inputs.emplace_back(TensorAdapter::AsGeTensorShared(item)); } vector ge_outputs; + ge_outputs.reserve(outputs.size()); for (auto &item : outputs) { - ge_outputs.emplace_back(TensorAdapter::AsGeTensor(item)); + ge_outputs.emplace_back(TensorAdapter::AsGeTensorShared(item)); } Status ret = graph_manager_.RunGraphWithStreamAsync(graph_id, stream, session_id_, ge_inputs, ge_outputs); domi::GetContext().out_nodes_map.clear();