From 293eea6bd53b4367bca4db69331789a2595057b6 Mon Sep 17 00:00:00 2001 From: guopeian Date: Sat, 26 Jun 2021 15:25:58 +0800 Subject: [PATCH] fix --- .../aicpu/aicpu_node_executor.cc | 176 +++++++++++++----- .../node_executor/aicpu/aicpu_node_executor.h | 89 +++++---- 2 files changed, 186 insertions(+), 79 deletions(-) diff --git a/ge/hybrid/node_executor/aicpu/aicpu_node_executor.cc b/ge/hybrid/node_executor/aicpu/aicpu_node_executor.cc index c83a76d1..96424cb6 100755 --- a/ge/hybrid/node_executor/aicpu/aicpu_node_executor.cc +++ b/ge/hybrid/node_executor/aicpu/aicpu_node_executor.cc @@ -230,7 +230,7 @@ Status AicpuNodeTaskBase::ExecuteAsync(TaskContext &context, std::functionnum_outputs == 0)) { GELOGD("Node[%s] type[%s] unknown_type is %d, output num is %d.", node_name_.c_str(), node_item_->node_type.c_str(), unknown_type_, node_item_->num_outputs); @@ -261,11 +261,12 @@ Status AicpuTfNodeTask::InitForDependComputeTask() { GE_CHK_STATUS_RET(AllocTensorBuffer(copy_input_buf_len, copy_input_dst_dev_), "[Alloc][TensorBuffer] failed for Node[%s] to copy task input dst, size=%zu", node_name_.c_str(), copy_input_buf_len); - // copy task args buf - GE_CHK_STATUS_RET(AllocTensorBuffer(sizeof(STR_FWK_OP_KERNEL), copy_task_args_buf_), - "[Alloc][TensorBuffer] failed for Node[%s] to copy task args, size=%zu", - node_name_.c_str(), sizeof(STR_FWK_OP_KERNEL)); + if (is_tfkernel) { + GE_CHK_STATUS_RET(AllocTensorBuffer(sizeof(STR_FWK_OP_KERNEL), copy_task_args_buf_), + "[Alloc][TensorBuffer] failed for Node[%s] to copy task args, size=%zu", + node_name_.c_str(), sizeof(STR_FWK_OP_KERNEL)); + } std::vector copy_io_addr; copy_io_addr.emplace_back(reinterpret_cast(copy_input_release_flag_dev_->GetData())); @@ -327,7 +328,7 @@ Status AicpuTfNodeTask::Init(const HybridModel &model) { uint64_t ext_session_id = model.GetSessionId(); GE_CHK_STATUS_RET(InitExtInfo(kernel_ext_info, ext_session_id), "[Init][ExtInfo] failed for Node[%s].", node_name_.c_str()); - GE_CHK_STATUS_RET(InitForDependComputeTask(), "[Init][DependComputeTask] failed for Node[%s].", node_name_.c_str()); + GE_CHK_STATUS_RET(InitForDependComputeTask(true), "[Init][DependComputeTask] failed for Node[%s].", node_name_.c_str()); // build fwk_op_kernel. GE_IF_BOOL_EXEC(sizeof(STR_FWK_OP_KERNEL) < kernel_ex_def.args_size(), @@ -441,8 +442,8 @@ Status AicpuTfNodeTask::EnsureSessionCreated(uint64_t session_id) { return SUCCESS; } -Status AicpuTfNodeTask::ReadResultSummaryAndPrepareMemory(TaskContext &context, - std::vector> &out_shape_hbm) { +Status AicpuNodeTaskBase::ReadResultSummaryAndPrepareMemory(TaskContext &context, + std::vector> &out_shape_hbm) { for (auto i = 0; i < node_item_->num_outputs; ++i) { auto &result_summary = output_summary_host_[i]; GE_CHK_RT_RET(rtMemcpy(&result_summary, sizeof(aicpu::FWKAdapter::ResultSummary), @@ -479,6 +480,7 @@ Status AicpuTfNodeTask::CopyDataToHbm(TaskContext &context, RECORD_CALLBACK_EVENT(context.GetExecutionContext(), node_name_.c_str(), "[LaunchCopy] Start"); GE_CHK_RT_RET(rtKernelLaunchEx(copy_task_args_buf_->GetData(), sizeof(STR_FWK_OP_KERNEL), RT_KERNEL_DEFAULT, context.GetStream())); + RECORD_CALLBACK_EVENT(context.GetExecutionContext(), node_name_.c_str(), "[LaunchCopy] End"); GE_CHK_RT_RET(rtStreamSynchronize(context.GetStream())); @@ -486,8 +488,33 @@ Status AicpuTfNodeTask::CopyDataToHbm(TaskContext &context, return SUCCESS; } -Status AicpuTfNodeTask::PrepareCopyInputs(const TaskContext &context, - const std::vector> &out_shape_hbm) { +Status AicpuNodeTask::CopyDataToHbm(TaskContext &context, + const std::vector> &out_shape_hbm) { + GE_CHK_BOOL_RET_STATUS(out_shape_hbm.size() == static_cast(node_item_->num_outputs), + INTERNAL_ERROR, + "[Check][Size]Node[%s] has %d outputs but out shape is %zu not equal.", + node_name_.c_str(), node_item_->num_outputs, out_shape_hbm.size()); + + GE_CHK_STATUS_RET_NOLOG(PrepareCopyInputs(context, out_shape_hbm)); + + RECORD_CALLBACK_EVENT(context.GetExecutionContext(), node_name_.c_str(), "[LaunchCopy] Start"); + uint32_t flag = RT_KERNEL_DEFAULT; + auto rt_ret = rtCpuKernelLaunchWithFlag(reinterpret_cast(memcpy_so_name.c_str()), + reinterpret_cast(memcpy_kernel_name.c_str()), + 1, // default core dim is 1 + memcpy_args_.get(), memcpy_args_size_, + nullptr, context.GetStream(), flag); + GE_CHK_RT_RET(rt_ret); + + RECORD_CALLBACK_EVENT(context.GetExecutionContext(), node_name_.c_str(), "[LaunchCopy] End"); + + GE_CHK_RT_RET(rtStreamSynchronize(context.GetStream())); + RECORD_CALLBACK_EVENT(context.GetExecutionContext(), node_name_.c_str(), "[SynchronizeCopy] End"); + return SUCCESS; +} + +Status AicpuNodeTaskBase::PrepareCopyInputs(const TaskContext &context, + const std::vector> &out_shape_hbm) { std::vector copy_input_release_flag; std::vector copy_input_data_size; std::vector copy_input_src; @@ -528,8 +555,8 @@ Status AicpuTfNodeTask::PrepareCopyInputs(const TaskContext &context, return SUCCESS; } -Status AicpuTfNodeTask::UpdateShapeByHbmBuffer(TaskContext &context, - const std::vector> &out_shape_hbm) { +Status AicpuNodeTaskBase::UpdateShapeByHbmBuffer(TaskContext &context, + const std::vector> &out_shape_hbm) { GE_CHK_BOOL_RET_STATUS(out_shape_hbm.size() == static_cast(node_item_->num_outputs), INTERNAL_ERROR, "Node[%s] has %d outputs but out shape is %zu", @@ -560,7 +587,7 @@ Status AicpuTfNodeTask::UpdateShapeByHbmBuffer(TaskContext &context, return SUCCESS; } -Status AicpuTfNodeTask::UpdateShapeAndDataByResultSummary(TaskContext &context) { +Status AicpuNodeTaskBase::UpdateShapeAndDataByResultSummary(TaskContext &context) { GELOGD("Node[%s] update shape and data by result summary begin.", node_name_.c_str()); std::vector> out_shape_hbm; @@ -648,7 +675,7 @@ Status AicpuTfNodeTask::LaunchTask(TaskContext &context) { return SUCCESS; } -Status AicpuTfNodeTask::TaskCallback(TaskContext &context) { +Status AicpuNodeTaskBase::TaskCallback(TaskContext &context) { GELOGD("Node[%s] task callback start. is_dynamic=%s, unknown_type=%d.", node_name_.c_str(), node_item_->is_dynamic ? "true" : "false", unknown_type_); Status callback_ret = SUCCESS; @@ -665,14 +692,71 @@ Status AicpuTfNodeTask::TaskCallback(TaskContext &context) { return callback_ret; } +Status AicpuNodeTask::SetMemCopyTask(const domi::TaskDef &task_def) { + if (node_item_->num_outputs == 0) { + GELOGD("Node[%s] type[%s] has no output, no need set mem_copy task.", + node_name_.c_str(), node_item_->node_type.c_str()); + return SUCCESS; + } + + GELOGD("Start to set memcpy task for node[%s].", node_name_.c_str()); + const domi::KernelDef &kernel_def = task_def.kernel(); + auto &memcpy_args = kernel_def.args(); + memcpy_args_size_ = kernel_def.args_size(); + memcpy_so_name_ = kernel_def.so_name(); + memcpy_kernel_name_ = kernel_def.kernel_name(); + GE_IF_BOOL_EXEC(memcpy_args.size() != memcpy_args_size_, + REPORT_INNER_ERROR("E19999", "MemCopy task def args.size=%zu, but args_size=%u not equal.", + memcpy_args.size(), memcpy_args_size_); + GELOGE(FAILED, "[Check][Size]MemCopy task def args.size=%zu, but args_size=%u not equal.", + memcpy_args.size(), memcpy_args_size_); + return FAILED;); + GE_IF_BOOL_EXEC(memcpy_args_size_ < sizeof(aicpu::AicpuParamHead), + REPORT_INNER_ERROR("E19999", + "Task def args_size=%u is less than aicpu param head len=%zu.", + memcpy_args_size_, sizeof(aicpu::AicpuParamHead)); + GELOGE(FAILED, + "[Check][Size] Task def args_size=%u is less than aicpu param head len=%zu.", + memcpy_args_size_, sizeof(aicpu::AicpuParamHead)); + return FAILED;); + + memcpy_args_.reset(new(std::nothrow) uint8_t[memcpy_args_size_]()); + GE_IF_BOOL_EXEC(memcpy_task == nullptr, + REPORT_INNER_ERROR("E19999", "new memory failed for Node[MemCopy], task_size[%u].", + memcpy_args_size_); + GELOGE(FAILED, "[Malloc][Memory] failed for Node[MemCopy], task_size[%u].", + memcpy_args_size_); + return FAILED;); + + errno_t sec_ret = memcpy_s(memcpy_args_.get(), memcpy_args_size_, memcpy_args.c_str(), memcpy_args.size()); + GE_IF_BOOL_EXEC(sec_ret != EOK, + REPORT_INNER_ERROR("E19999", + "memcpy_s argc_ failed for Node[MemCopy], ret: %d", sec_ret); + GELOGE(INTERNAL_ERROR, + "[Update][args] failed for Node[MemCopy], ret: %d", sec_ret); + return sec_ret;); + auto memcpy_param_head = reinterpret_cast(memcpy_args_.get()); + uint32_t memcpy_io_num = memcpy_param_head->ioAddrNum; + auto memcpy_io_addr = memcpy_args_.get() + sizeof(aicpu::AicpuParamHead); + // if has input and output, need copy to ioaddr + int cpy_ret = memcpy_s(memcpy_io_addr, memcpy_args_size_ - sizeof(aicpu::AicpuParamHead), + ©_ioaddr_dev_, sizeof(uint64_t) * memcpy_io_num); + GE_IF_BOOL_EXEC(cpy_ret != 0, + REPORT_INNER_ERROR("E19999", "Node[Memcpoy] memcpy io addr to AicpuParamHead failed," + "ret=%d, args_size=%u, io nums=%zu.", + cpy_ret, memcpy_args_size_, memcpy_io_num); + GELOGE(INTERNAL_ERROR, "[Update][io_addr]Node[MemCopy] memcpy io addr to AicpuParamHead failed," + "ret=%d, args_size=%u, io nums=%zu.", + cpy_ret, memcpy_args_size_, memcpy_io_num); + return INTERNAL_ERROR;); + GELOGD("Set memcpy task for node[MemCopy] successfully."); + return SUCCESS; +} + Status AicpuNodeTask::Init(const HybridModel &model) { auto node_name = node_name_; GELOGD("Node[%s] init start.", node_name.c_str()); - GE_CHK_BOOL_RET_STATUS(unknown_type_ != DEPEND_COMPUTE, FAILED, - "[Check][Type]Node[%s] unknown type[%d] is depend compute, it's not supported now.", - node_name.c_str(), unknown_type_); - GE_CHK_BOOL_RET_STATUS(task_def_.has_kernel(), FAILED, "[Check][task_def_]Node[%s] task def does not has kernel.", node_name.c_str()); auto &kernel_def = task_def_.kernel(); @@ -762,6 +846,10 @@ Status AicpuNodeTask::Init(const HybridModel &model) { GE_CHK_STATUS_RET(InitExtInfo(kernel_ext_info, ext_session_id), "[Init][ExtInfo] failed for Node[%s].", node_name.c_str()); + GE_CHK_STATUS_RET(InitForDependComputeTask(false), + "[Init][DependComputeTask] failed for Node[%s].", + node_name_.c_str()); + if (ext_info_addr_dev_ == nullptr) { aicpu_param_head->extInfoLength = 0; aicpu_param_head->extInfoAddr = 0; @@ -770,6 +858,11 @@ Status AicpuNodeTask::Init(const HybridModel &model) { aicpu_param_head->extInfoAddr = reinterpret_cast(ext_info_addr_dev_->GetData()); } + auto task_defs = model.GetTaskDefs(node_item_->node); + GE_CHECK_NOTNULL(task_defs); + if (unknown_type_ == DEPEND_COMPUTE) { + GE_CHK_STATUS_RET_NOLOG(SetMemCopyTask((*task_defs)[1])); + } GELOGD("Node[%s] init end.", node_name.c_str()); return SUCCESS; } @@ -784,14 +877,30 @@ Status AicpuNodeTask::UpdateIoAddr(TaskContext &context) { GELOGD("Node[%s] input[%d] = %p, size = %zu", node_name_.c_str(), i, inputData->GetData(), inputData->GetSize()); io_addrs.emplace_back(reinterpret_cast(inputData->GetData())); } + // known shape or not depend compute + if (!node_item_->is_dynamic || unknown_type_ != DEPEND_COMPUTE) { + // unknown type 4 do this in call back. + GE_CHK_STATUS_RET_NOLOG(context.AllocateOutputs()); + for (auto j = 0; j < node_item_->num_outputs; ++j) { + auto outputData = context.GetOutput(j); + GE_CHECK_NOTNULL(outputData); - GE_CHK_STATUS_RET_NOLOG(context.AllocateOutputs()); - for (auto j = 0; j < node_item_->num_outputs; ++j) { - auto outputData = context.GetOutput(j); - GE_CHECK_NOTNULL(outputData); - GELOGD("Node[%s] output[%d] addr = %p, size = %zu", node_name_.c_str(), j, - outputData->GetData(), outputData->GetSize()); - io_addrs.emplace_back(reinterpret_cast(outputData->GetData())); + GELOGD("Node[%s] output[%d] addr = %p, size = %zu", + node_name_.c_str(), j, outputData->GetData(), outputData->GetSize()); + io_addrs.emplace_back(reinterpret_cast(outputData->GetData())); + } + } else { + // unknown type 4 use result summary update ioaddr. + GELOGD("Node[%s] is depend compute node, use result summary as out addr.", node_name_.c_str()); + GE_CHK_BOOL_RET_STATUS(output_summary_.size() == static_cast(node_item_->num_outputs), + INTERNAL_ERROR, + "[Check][Size]Node[%s] has %d output but %zu output summary not equal.", + node_name_.c_str(), node_item_->num_outputs, output_summary_.size()); + + for (auto j = 0; j < node_item_->num_outputs; ++j) { + void *summary_addr = output_summary_[j]->GetData(); + io_addrs.emplace_back(reinterpret_cast(summary_addr)); + } } auto io_addr = args_.get() + sizeof(aicpu::AicpuParamHead); @@ -829,23 +938,6 @@ Status AicpuNodeTask::LaunchTask(TaskContext &context) { return SUCCESS; } -Status AicpuNodeTask::TaskCallback(TaskContext &context) { - GELOGD("Node[%s] task callback start, is_dynamic = %s, unknown_type=%d.", - node_name_.c_str(), node_item_->is_dynamic ? "true" : "false", unknown_type_); - Status callback_ret = SUCCESS; - - // check need update shape, call update shape. - if (node_item_->is_dynamic && unknown_type_ == DEPEND_SHAPE_RANGE) { - // check result - callback_ret = UpdateOutputShapeFromExtInfo(context); - } else { - GELOGD("Node[%s] unknown shape type is %d no need update output shape.", - node_name_.c_str(), unknown_type_); - } - GELOGD("Node[%s] task callback end.", node_name_.c_str()); - return callback_ret; -} - Status AiCpuNodeExecutor::PrepareTask(NodeTask &task, TaskContext &context) const { // malloc HBM memory at Init, here just update them RECORD_EXECUTION_EVENT(context.GetExecutionContext(), context.GetNodeName(), "[AiCpuNodeExecutorPrepareTask] Start"); diff --git a/ge/hybrid/node_executor/aicpu/aicpu_node_executor.h b/ge/hybrid/node_executor/aicpu/aicpu_node_executor.h index 14bc8fcc..2bfdd615 100644 --- a/ge/hybrid/node_executor/aicpu/aicpu_node_executor.h +++ b/ge/hybrid/node_executor/aicpu/aicpu_node_executor.h @@ -54,13 +54,37 @@ class AicpuNodeTaskBase : public NodeTask { Status UpdateShapeToOutputDesc(TaskContext &task_context, const GeShape &shape_new, int32_t output_index); virtual Status LaunchTask(TaskContext &context) = 0; + + virtual Status UpdateShapeAndDataByResultSummary(TaskContext &context); - virtual Status TaskCallback(TaskContext &context) = 0; + Status InitForDependComputeTask(bool is_tfkernel); + + Status TaskCallback(TaskContext &context); virtual Status UpdateIoAddr(TaskContext &context) = 0; static Status AllocTensorBuffer(size_t size, std::unique_ptr &tensor_buffer); + virtual Status CopyDataToHbm(TaskContext &context, + const std::vector> &out_shape_hbm) = 0; +private: + Status TaskCallback(TaskContext &context); + + /// + /// read result summary and prepare copy task memory. + /// @param context task context + /// @param out_shape_hbm if scalar, TensorBuffer->data is null, size=0 + /// @return SUCCESS:success other:failed + /// + Status ReadResultSummaryAndPrepareMemory(TaskContext &context, + std::vector> &out_shape_hbm); + + Status UpdateShapeByHbmBuffer(TaskContext &context, + const std::vector> &out_shape_hbm); + + Status PrepareCopyInputs(const TaskContext &context, + const std::vector> &out_shape_hbm); + protected: const NodeItem *node_item_; // just reference. @@ -78,6 +102,17 @@ class AicpuNodeTaskBase : public NodeTask { // ext info addr, device mem std::unique_ptr ext_info_addr_dev_; + + std::vector> output_summary_; + std::vector output_summary_host_; + + std::unique_ptr copy_ioaddr_dev_; + + std::unique_ptr copy_input_release_flag_dev_; + std::unique_ptr copy_input_data_size_dev_; + std::unique_ptr copy_input_src_dev_; + std::unique_ptr copy_input_dst_dev_; + bool need_sync_ = false; }; class AicpuTfNodeTask : public AicpuNodeTaskBase { @@ -93,34 +128,13 @@ class AicpuTfNodeTask : public AicpuNodeTaskBase { Status LaunchTask(TaskContext &context) override; - Status TaskCallback(TaskContext &context) override; - Status UpdateIoAddr(TaskContext &context) override; - + + Status CopyDataToHbm(TaskContext &context, + const std::vector> &out_shape_hbm) override; private: Status SetMemCopyTask(const domi::TaskDef &task_def); - Status InitForDependComputeTask(); - - Status UpdateShapeAndDataByResultSummary(TaskContext &context); - - /// - /// read result summary and prepare copy task memory. - /// @param context task context - /// @param out_shape_hbm if scalar, TensorBuffer->data is null, size=0 - /// @return SUCCESS:success other:failed - /// - Status ReadResultSummaryAndPrepareMemory(TaskContext &context, - std::vector> &out_shape_hbm); - Status CopyDataToHbm(TaskContext &context, - const std::vector> &out_shape_hbm); - - Status UpdateShapeByHbmBuffer(TaskContext &context, - const std::vector> &out_shape_hbm); - - Status PrepareCopyInputs(const TaskContext &context, - const std::vector> &out_shape_hbm); - static Status EnsureSessionCreated(uint64_t session_id); static uint64_t GetStepIdAddr(const HybridModel &model); private: @@ -135,17 +149,6 @@ class AicpuTfNodeTask : public AicpuNodeTaskBase { // just used for depend DEPEND_COMPUTE op std::unique_ptr copy_task_args_buf_; - std::vector> output_summary_; - std::vector output_summary_host_; - - std::unique_ptr copy_ioaddr_dev_; - - std::unique_ptr copy_input_release_flag_dev_; - std::unique_ptr copy_input_data_size_dev_; - std::unique_ptr copy_input_src_dev_; - std::unique_ptr copy_input_dst_dev_; - bool need_sync_ = false; - std::unique_ptr copy_workspace_buf_; }; @@ -162,16 +165,28 @@ class AicpuNodeTask : public AicpuNodeTaskBase { Status LaunchTask(TaskContext &context) override; - Status TaskCallback(TaskContext &context) override; + Status CopyDataToHbm(TaskContext &context, + const std::vector> &out_shape_hbm) override; Status UpdateIoAddr(TaskContext &context) override; +private: + Status SetMemCopyTask(const domi::TaskDef &task_def); + protected: // host mem std::unique_ptr args_; + // host memcpy mem + std::unique_ptr memcpy_args_; // args size uint32_t args_size_ = 0; + + std::string memcpy_so_name_; + + std::string memcpy_kernel_name_; + // args size + uint32_t memcpy_args_size_ = 0; }; class AiCpuNodeExecutor : public NodeExecutor {