!1598 parse topic type

From: @li-lei0106
Reviewed-by: @wqtshg
Signed-off-by:

ge/graph/load/model_manager/task_info/kernel_ex_task_info.cc

@@ -53,6 +53,7 @@ Status KernelExTaskInfo::InitTaskExtInfo(const std::string &ext_info, const OpDe
                     "Parse kernel ext info failed, kernel_ext_info_size=%zu.", ext_info.size());
   GE_CHK_STATUS_RET(ext_handle->UpdateExecuteMode(true), "UpdateExecuteMode failed.");
   GELOGD("Update aicpu_task ext_info bit_map execute mode to 1.");
+  topic_type_flag_ = ext_handle->GetTopicTypeFlag();
 
   bool all_shape = false;
   (void)AttrUtils::GetBool(op_desc, kAicpuAllshape, all_shape);
@@ -407,6 +408,14 @@ Status KernelExTaskInfo::CopyTaskInfo(const domi::KernelExDef &kernel_def, const
 
 Status KernelExTaskInfo::Distribute() {
   GELOGI("KernelExTaskInfo Distribute Start.");
+  // Use the fifth and sixth bits of dump_flag_ indicate the value of topic_type.
+  // xxxxxxxx xxxxxxxx xxxxxxxx xx00xxxx: DEVICE_ONLY
+  // xxxxxxxx xxxxxxxx xxxxxxxx xx01xxxx: DEVICE_FIRST
+  // xxxxxxxx xxxxxxxx xxxxxxxx xx10xxxx: HOST_ONLY
+  // xxxxxxxx xxxxxxxx xxxxxxxx xx11xxxx: HOST_FIRST
+  if (topic_type_flag_ > 0) {
+    dump_flag_ = dump_flag_ | topic_type_flag_;
+  }
   rtError_t rt_ret = rtKernelLaunchEx(kernel_buf_, kernel_buf_size_, dump_flag_, stream_);
   if (rt_ret != RT_ERROR_NONE) {
     REPORT_CALL_ERROR("E19999", "Call rtKernelLaunchEx failed, ret:0x%X",

ge/graph/load/model_manager/task_info/kernel_ex_task_info.h

@@ -76,6 +76,7 @@ class KernelExTaskInfo : public TaskInfo {
   vector<void *> io_addrs_;
   uint32_t args_offset_ = 0;
   int64_t fixed_addr_offset_ = 0;
+  int32_t topic_type_flag_ = -1;
 };
 }  // namespace ge
 #endif  // GE_GRAPH_LOAD_NEW_MODEL_MANAGER_TASK_INFO_KERNEL_EX_TASK_INFO_H_

ge/graph/load/model_manager/task_info/kernel_task_info.cc

@@ -431,6 +431,14 @@ Status KernelTaskInfo::Distribute() {
   int64_t env_flag = (res == EN_OK) ? strtol(skt_enable_env, nullptr, kBaseInt) : kStrtolFail;
   bool call_skt = ((env_flag != 0) || is_l1_fusion_enable_);
   if (kernel_type_ == ccKernelType::AI_CPU || kernel_type_ == ccKernelType::CUST_AI_CPU) {
+    if (topic_type_flag_ > 0) {
+      // Use the fifth and sixth bits of dump_flag_ indicate the value of topic_type.
+      // xxxxxxxx xxxxxxxx xxxxxxxx xx00xxxx: DEVICE_ONLY
+      // xxxxxxxx xxxxxxxx xxxxxxxx xx01xxxx: DEVICE_FIRST
+      // xxxxxxxx xxxxxxxx xxxxxxxx xx10xxxx: HOST_ONLY
+      // xxxxxxxx xxxxxxxx xxxxxxxx xx11xxxx: HOST_FIRST
+      dump_flag_ = dump_flag_ | topic_type_flag_;
+    }
     GELOGI("distribute task info kernel_type %d, flag %d", kernel_type_, dump_flag_);
     // blockDim is reserved parameter, set to 1
     rt_ret = rtCpuKernelLaunchWithFlag(reinterpret_cast<const void *>(so_name_.c_str()),
@@ -1116,6 +1124,7 @@ Status KernelTaskInfo::InitAicpuTaskExtInfo(const std::string &ext_info) {
   GELOGD("Update aicpu_task ext_info session_info session_id is %lu", davinci_model_->GetSessionId());
   GE_CHK_STATUS_RET(ext_handle->UpdateExecuteMode(true), "UpdateExecuteMode failed.");
   GELOGD("Update aicpu_task ext_info bit_map execute mode to 1.");
+  topic_type_flag_ = ext_handle->GetTopicTypeFlag();
 
   bool all_shape = false;
   (void)AttrUtils::GetBool(op_desc_, kAicpuAllshape, all_shape);

ge/graph/load/model_manager/task_info/kernel_task_info.h

@@ -169,6 +169,7 @@ class KernelTaskInfo : public TaskInfo {
   uint16_t io_addr_offset_ = 0;
   bool l2_buffer_on_ = false;
   bool call_save_dump_ = false;
+  int32_t topic_type_flag_ = -1;
 
   // aicpu ext_info device mem
   void *aicpu_ext_info_addr_ = nullptr;

ge/hybrid/node_executor/aicpu/aicpu_ext_info.cc

@@ -24,6 +24,12 @@ namespace hybrid {
 namespace {
 // if dim count is not reach kMaxShapeDims(8), use INT64_MIN to mark dim end.
 constexpr int64_t kDimEndFlag = INT64_MIN;
+const std::map<int32_t, int32_t> kTopicTypeToRtsFlagMap {
+    {static_cast<int32_t>(aicpu::FWKAdapter::FWK_ADPT_TOPIC_DEVICE_ONLY), 0},
+    {static_cast<int32_t>(aicpu::FWKAdapter::FWK_ADPT_TOPIC_DEVICE_FIRST), RT_KERNEL_DEVICE_FIRST},
+    {static_cast<int32_t>(aicpu::FWKAdapter::FWK_ADPT_TOPIC_HOST_ONLY), RT_KERNEL_HOST_ONLY},
+    {static_cast<int32_t>(aicpu::FWKAdapter::FWK_ADPT_TOPIC_HOST_FIRST), RT_KERNEL_HOST_FIRST}
+};
 }
 
 Status AicpuExtInfoHandler::Parse(const std::string &ext_info) {
@@ -72,6 +78,9 @@ Status AicpuExtInfoHandler::Parse(const std::string &ext_info) {
       case aicpu::FWKAdapter::FWK_ADPT_EXT_UPDATE_ADDR:
         GE_CHK_STATUS_RET(ParseExtUpdateAddr(aicpu_ext_info), "[Parse][ExtUpdateAddr] failed.");
         break;
+      case aicpu::FWKAdapter::FWK_ADPT_EXT_TOPIC_TYPE:
+        GE_CHK_STATUS_RET(ParseExtTopicType(aicpu_ext_info), "[Parse][ExtTopicType] failed.");
+        break;
       default:
         GELOGD("Node[%s] ignore infoType=%d, infoLen=%u.",
                node_name_.c_str(), aicpu_ext_info->infoType, aicpu_ext_info->infoLen);
@@ -207,6 +216,44 @@ Status AicpuExtInfoHandler::ParseExtUpdateAddr(AicpuExtInfo *aicpu_ext_info) {
   return SUCCESS;
 }
 
+Status AicpuExtInfoHandler::ParseExtTopicType(AicpuExtInfo *aicpu_ext_info) {
+  if (aicpu_ext_info->infoLen != sizeof(int32_t)) {
+    REPORT_INNER_ERROR("E19999",
+                       "Node[%s] parse topic_type info failed as infoLen must be %zu but %u.",
+                       node_name_.c_str(), sizeof(int32_t), aicpu_ext_info->infoLen);
+    GELOGE(ACL_ERROR_GE_PARAM_INVALID,
+           "[Check][DataLen]Node[%s] parse topic_type info failed as infoLen must be %zu but %u.",
+           node_name_.c_str(), sizeof(int32_t), aicpu_ext_info->infoLen);
+    return ACL_ERROR_GE_PARAM_INVALID;
+  }
+  GE_CHECK_NOTNULL(aicpu_ext_info->infoMsg);
+  auto type = *reinterpret_cast<const int32_t *>(aicpu_ext_info->infoMsg);
+
+  topic_type_flag_ = TopicTypeToRtsFlag(type);
+  if (topic_type_flag_ == -1) {
+    REPORT_INNER_ERROR("E19999", "Node[%s] parse ext topic type failed as need %d %d %d %d but %d.",
+                       node_name_.c_str(),
+                       aicpu::FWKAdapter::FWK_ADPT_TOPIC_DEVICE_ONLY,
+                       aicpu::FWKAdapter::FWK_ADPT_TOPIC_DEVICE_FIRST,
+                       aicpu::FWKAdapter::FWK_ADPT_TOPIC_HOST_ONLY,
+                       aicpu::FWKAdapter::FWK_ADPT_TOPIC_HOST_FIRST,
+                       type);
+    GELOGE(ACL_ERROR_GE_PARAM_INVALID,
+           "[Check][Type]Node[%s] parse ext shape type failed as need %d %d %d %d but %d.",
+           node_name_.c_str(),
+           aicpu::FWKAdapter::FWK_ADPT_TOPIC_DEVICE_ONLY,
+           aicpu::FWKAdapter::FWK_ADPT_TOPIC_DEVICE_FIRST,
+           aicpu::FWKAdapter::FWK_ADPT_TOPIC_HOST_ONLY,
+           aicpu::FWKAdapter::FWK_ADPT_TOPIC_HOST_FIRST,
+           type);
+    return ACL_ERROR_GE_PARAM_INVALID;
+  }
+
+  GELOGI("Node[%s] parse ext topic type info success infoLen=%u, topic_type=%d, rts_flag=%d.",
+         node_name_.c_str(), aicpu_ext_info->infoLen, type, topic_type_flag_);
+  return SUCCESS;
+}
+
 Status AicpuExtInfoHandler::UpdateExecuteMode(bool flag) {
   if (bit_map_ == nullptr) {
     GELOGD("There is no bit_map in ext_info, no need update.");
@@ -341,5 +388,14 @@ void AicpuExtInfoHandler::GetShapeAndType(const AicpuShapeAndType *shape_and_typ
   data_type = static_cast<DataType>(shape_and_type->type);
   shape = GeShape(dims);
 }
+
+int32_t AicpuExtInfoHandler::TopicTypeToRtsFlag(int32_t topic_type) {
+  auto it = kTopicTypeToRtsFlagMap.find(topic_type);
+  if (it != kTopicTypeToRtsFlagMap.end()) {
+    return it->second;
+  }
+
+  return -1;
+}
 }  // namespace hybrid
 }  // namespace ge

ge/hybrid/node_executor/aicpu/aicpu_ext_info.h

@@ -62,6 +62,7 @@ class AicpuExtInfoHandler {
   Status GetOutputShapeAndType(uint32_t output_index, GeShape &shape, DataType &data_type);
 
   bool IsNeedRefreshIOAddr();
+  int32_t GetTopicTypeFlag() const { return topic_type_flag_; }
 
  private:
 
@@ -71,6 +72,7 @@ class AicpuExtInfoHandler {
   Status ParseExtSessionInfo(AicpuExtInfo *aicpu_ext_info);
   Status ParseExtBitMap(AicpuExtInfo *aicpu_ext_info);
   Status ParseExtUpdateAddr(AicpuExtInfo *aicpu_ext_info);
+  Status ParseExtTopicType(AicpuExtInfo *aicpu_ext_info);
 
   static Status UpdateShapeAndType(const GeShape &shape,
                                    DataType data_type,
@@ -81,6 +83,8 @@ class AicpuExtInfoHandler {
                               DataType &data_type);
 
  private:
+  int32_t TopicTypeToRtsFlag(int32_t topic_type);
+
   const std::string node_name_;
   const uint32_t input_num_;
   const uint32_t output_num_;
@@ -88,6 +92,7 @@ class AicpuExtInfoHandler {
   AicpuSessionInfo *session_info_ = nullptr;
   uint64_t *bit_map_ = nullptr;
   uint32_t *update_addr_ = nullptr;
+  int32_t topic_type_flag_ = -1;
 
   std::unique_ptr<uint8_t[]> ext_info_;
   size_t ext_info_len_ = 0;

tests/ut/ge/graph/load/kernel_ex_task_info_unittest.cc

@@ -155,4 +155,58 @@ TEST_F(UtestKernelExTaskInfo, parse_update_addr) {
   KernelExTaskInfo kernel_ex_task_info;
   EXPECT_EQ(kernel_ex_task_info.InitTaskExtInfo(ext_info, op_desc), SUCCESS);
 }
+
+TEST_F(UtestKernelExTaskInfo, parse_topic_type_success_1) {
+  const string ext_info = {7,0,0,0,4,0,0,0,0,0,0,0};
+  const OpDescPtr op_desc = CreateOpDesc("FrameworkOp", "FrameworkOp");
+  AttrUtils::SetBool(op_desc, "_AllShape", true);
+
+  KernelExTaskInfo kernel_ex_task_info;
+  EXPECT_EQ(kernel_ex_task_info.InitTaskExtInfo(ext_info, op_desc), SUCCESS);
+}
+
+TEST_F(UtestKernelExTaskInfo, parse_topic_type_success_2) {
+  const string ext_info = {7,0,0,0,4,0,0,0,1,0,0,0};
+  const OpDescPtr op_desc = CreateOpDesc("FrameworkOp", "FrameworkOp");
+  AttrUtils::SetBool(op_desc, "_AllShape", true);
+
+  KernelExTaskInfo kernel_ex_task_info;
+  EXPECT_EQ(kernel_ex_task_info.InitTaskExtInfo(ext_info, op_desc), SUCCESS);
+}
+
+TEST_F(UtestKernelExTaskInfo, parse_topic_type_success_3) {
+  const string ext_info = {7,0,0,0,4,0,0,0,2,0,0,0};
+  const OpDescPtr op_desc = CreateOpDesc("FrameworkOp", "FrameworkOp");
+  AttrUtils::SetBool(op_desc, "_AllShape", true);
+
+  KernelExTaskInfo kernel_ex_task_info;
+  EXPECT_EQ(kernel_ex_task_info.InitTaskExtInfo(ext_info, op_desc), SUCCESS);
+}
+
+TEST_F(UtestKernelExTaskInfo, parse_topic_type_success_4) {
+  const string ext_info = {7,0,0,0,4,0,0,0,3,0,0,0};
+  const OpDescPtr op_desc = CreateOpDesc("FrameworkOp", "FrameworkOp");
+  AttrUtils::SetBool(op_desc, "_AllShape", true);
+
+  KernelExTaskInfo kernel_ex_task_info;
+  EXPECT_EQ(kernel_ex_task_info.InitTaskExtInfo(ext_info, op_desc), SUCCESS);
+}
+
+TEST_F(UtestKernelExTaskInfo, parse_topic_type_failed_1) {
+  const string ext_info = {7,0,0,0,4,0,0,0,4,0,0,0};
+  const OpDescPtr op_desc = CreateOpDesc("FrameworkOp", "FrameworkOp");
+  AttrUtils::SetBool(op_desc, "_AllShape", true);
+
+  KernelExTaskInfo kernel_ex_task_info;
+  EXPECT_NE(kernel_ex_task_info.InitTaskExtInfo(ext_info, op_desc), SUCCESS);
+}
+
+TEST_F(UtestKernelExTaskInfo, parse_topic_type_failed_2) {
+  const string ext_info = {7,0,0,0,2,0,0,0,2,0,0,0};
+  const OpDescPtr op_desc = CreateOpDesc("FrameworkOp", "FrameworkOp");
+  AttrUtils::SetBool(op_desc, "_AllShape", true);
+
+  KernelExTaskInfo kernel_ex_task_info;
+  EXPECT_NE(kernel_ex_task_info.InitTaskExtInfo(ext_info, op_desc), SUCCESS);
+}
 }  // namespace ge

third_party/fwkacllib/inc/cce/fwk_adpt_struct.h

@@ -61,9 +61,18 @@ enum FWKTaskExtInfoType {
   FWK_ADPT_EXT_OP_NAME,
   FWK_ADPT_EXT_SESSION_INFO,
   FWK_ADPT_EXT_BITMAP,
+  FWK_ADPT_EXT_TOPIC_TYPE,
   FWK_ADPT_EXT_INVALID
 };
 
+enum FWKExtTopicType {
+  FWK_ADPT_TOPIC_DEVICE_ONLY = 0,
+  FWK_ADPT_TOPIC_DEVICE_FIRST,
+  FWK_ADPT_TOPIC_HOST_ONLY,
+  FWK_ADPT_TOPIC_HOST_FIRST,
+  FWK_ADPT_TOPIC_INVALID
+};
+
 enum FWKExtUpdateAddrType {
   FWK_ADPT_UPDATE_NULL = 0,
   FWK_ADPT_UPDATE_INPUT,

third_party/fwkacllib/inc/runtime/kernel.h

@@ -191,6 +191,11 @@ typedef void (*rtCallback_t)(void *fnData);
 #define RT_FUSION_KERNEL_DUMPFLAG (0x04)
 #define RT_KERNEL_CUSTOM_AICPU (0x08)
 
+// STARS topic scheduler sqe : topic_type
+#define RT_KERNEL_DEVICE_FIRST (0X10)
+#define RT_KERNEL_HOST_ONLY (0X20)
+#define RT_KERNEL_HOST_FIRST (0X30)
+
 /**
  * @ingroup rt_kernel
  * @brief kernel mode