hummingbird
/
graphengine
Not watched
2
0
Fork 0
Code
Releases 13 Wiki Activity
Issues 0 Pull Requests 0 Datasets Model Cloudbrain
Browse Source

fix

pull/1879/head
guopeian 4 years ago
parent
5f2e411185
commit
293eea6bd5
2 changed files with 186 additions and 79 deletions
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. +134
    -42
      ge/hybrid/node_executor/aicpu/aicpu_node_executor.cc
  2. +52
    -37
      ge/hybrid/node_executor/aicpu/aicpu_node_executor.h

+ 134
- 42
ge/hybrid/node_executor/aicpu/aicpu_node_executor.cc View File

@@ -230,7 +230,7 @@ Status AicpuNodeTaskBase::ExecuteAsync(TaskContext &context, std::function<void(
return SUCCESS;
}

Status AicpuTfNodeTask::InitForDependComputeTask() {
Status AicpuNodeTaskBase::InitForDependComputeTask(bool is_tfkernel) {
if ((unknown_type_ != DEPEND_COMPUTE) || (node_item_->num_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<uint64_t> copy_io_addr;
copy_io_addr.emplace_back(reinterpret_cast<uintptr_t>(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<std::unique_ptr<TensorBuffer>> &out_shape_hbm) {
Status AicpuNodeTaskBase::ReadResultSummaryAndPrepareMemory(TaskContext &context,
std::vector<std::unique_ptr<TensorBuffer>> &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<std::unique_ptr<TensorBuffer>> &out_shape_hbm) {
Status AicpuNodeTask::CopyDataToHbm(TaskContext &context,
const std::vector<std::unique_ptr<TensorBuffer>> &out_shape_hbm) {
GE_CHK_BOOL_RET_STATUS(out_shape_hbm.size() == static_cast<std::size_t>(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<const void *>(memcpy_so_name.c_str()),
reinterpret_cast<const void *>(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<std::unique_ptr<TensorBuffer>> &out_shape_hbm) {
std::vector<uint64_t> copy_input_release_flag;
std::vector<uint64_t> copy_input_data_size;
std::vector<uint64_t> copy_input_src;
@@ -528,8 +555,8 @@ Status AicpuTfNodeTask::PrepareCopyInputs(const TaskContext &context,
return SUCCESS;
}

Status AicpuTfNodeTask::UpdateShapeByHbmBuffer(TaskContext &context,
const std::vector<std::unique_ptr<TensorBuffer>> &out_shape_hbm) {
Status AicpuNodeTaskBase::UpdateShapeByHbmBuffer(TaskContext &context,
const std::vector<std::unique_ptr<TensorBuffer>> &out_shape_hbm) {
GE_CHK_BOOL_RET_STATUS(out_shape_hbm.size() == static_cast<std::size_t>(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<std::unique_ptr<TensorBuffer>> 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<aicpu::AicpuParamHead *>(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),
&copy_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<uintptr_t>(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<uintptr_t>(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<uintptr_t>(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<uintptr_t>(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<std::size_t>(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<uintptr_t>(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");


+ 52
- 37
ge/hybrid/node_executor/aicpu/aicpu_node_executor.h View File

@@ -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<TensorBuffer> &tensor_buffer);

virtual Status CopyDataToHbm(TaskContext &context,
const std::vector<std::unique_ptr<TensorBuffer>> &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<std::unique_ptr<TensorBuffer>> &out_shape_hbm);

Status UpdateShapeByHbmBuffer(TaskContext &context,
const std::vector<std::unique_ptr<TensorBuffer>> &out_shape_hbm);

Status PrepareCopyInputs(const TaskContext &context,
const std::vector<std::unique_ptr<TensorBuffer>> &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<TensorBuffer> ext_info_addr_dev_;

std::vector<std::unique_ptr<TensorBuffer>> output_summary_;
std::vector<aicpu::FWKAdapter::ResultSummary> output_summary_host_;

std::unique_ptr<TensorBuffer> copy_ioaddr_dev_;

std::unique_ptr<TensorBuffer> copy_input_release_flag_dev_;
std::unique_ptr<TensorBuffer> copy_input_data_size_dev_;
std::unique_ptr<TensorBuffer> copy_input_src_dev_;
std::unique_ptr<TensorBuffer> 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<std::unique_ptr<TensorBuffer>> &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<std::unique_ptr<TensorBuffer>> &out_shape_hbm);
Status CopyDataToHbm(TaskContext &context,
const std::vector<std::unique_ptr<TensorBuffer>> &out_shape_hbm);

Status UpdateShapeByHbmBuffer(TaskContext &context,
const std::vector<std::unique_ptr<TensorBuffer>> &out_shape_hbm);

Status PrepareCopyInputs(const TaskContext &context,
const std::vector<std::unique_ptr<TensorBuffer>> &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<TensorBuffer> copy_task_args_buf_;

std::vector<std::unique_ptr<TensorBuffer>> output_summary_;
std::vector<aicpu::FWKAdapter::ResultSummary> output_summary_host_;

std::unique_ptr<TensorBuffer> copy_ioaddr_dev_;

std::unique_ptr<TensorBuffer> copy_input_release_flag_dev_;
std::unique_ptr<TensorBuffer> copy_input_data_size_dev_;
std::unique_ptr<TensorBuffer> copy_input_src_dev_;
std::unique_ptr<TensorBuffer> copy_input_dst_dev_;
bool need_sync_ = false;

std::unique_ptr<TensorBuffer> 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<std::unique_ptr<TensorBuffer>> &out_shape_hbm) override;

Status UpdateIoAddr(TaskContext &context) override;

private:
Status SetMemCopyTask(const domi::TaskDef &task_def);

protected:
// host mem
std::unique_ptr<uint8_t[]> args_;

// host memcpy mem
std::unique_ptr<uint8_t[]> 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 {


Write Preview
Loading…
Cancel
Save