Line data    Source code

       1             : /**
       2             :  * Copyright (c) 2025 Huawei Technologies Co., Ltd.
       3             :  * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
       4             :  * CANN Open Software License Agreement Version 2.0 (the "License").
       5             :  * Please refer to the License for details. You may not use this file except in compliance with the License.
       6             :  * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
       7             :  * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
       8             :  * See LICENSE in the root of the software repository for the full text of the License.
       9             :  */
      10             : #include "queue_process_ccpu.h"
      11             : #include "log_inner.h"
      12             : #include "runtime/rt_mem_queue.h"
      13             : #include "runtime/dev.h"
      14             : #include "queue_schedule/qs_client.h"
      15             : 
      16             : namespace acl {
      17           0 :     aclError QueueProcessorCcpu::acltdtCreateQueue(const acltdtQueueAttr *const attr, uint32_t *const qid)
      18             :     {
      19           0 :         ACL_LOG_INFO("Start to create queue");
      20           0 :         ACL_REQUIRES_NOT_NULL(qid);
      21           0 :         ACL_REQUIRES_OK(acltdtCreateGroup());
      22           0 :         constexpr int32_t deviceId = 0;
      23             :         static bool isQueueIint = false;
      24           0 :         if (!isQueueIint) {
      25           0 :             ACL_LOG_INFO("need to init queue once");
      26           0 :             const rtError_t ret =  rtMemQueueInit(deviceId);
      27           0 :             if ((ret != ACL_RT_SUCCESS) && (ret != ACL_ERROR_RT_REPEATED_INIT)) {
      28           0 :                 ACL_LOG_INNER_ERROR("queue init failed, ret is %d", ret);
      29           0 :                 return ret;
      30             :             }
      31           0 :             isQueueIint = true;
      32             :         }
      33           0 :         ACL_REQUIRES_OK(acltdtCreateQueueWithAttr(deviceId, attr, qid));
      34           0 :         ACL_LOG_INFO("Successfully to execute create queue, qid is %u", *qid);
      35           0 :         return ACL_SUCCESS;
      36             :     }
      37             : 
      38           1 :     aclError QueueProcessorCcpu::acltdtCreateGroup()
      39             :     {
      40             :         static bool isGroupCreate = false;
      41           1 :         if (isGroupCreate) {
      42           0 :             return ACL_SUCCESS;
      43             :         }
      44           1 :         ACL_LOG_INFO("Start to create group");
      45           1 :         size_t grpNum = 0U;
      46           2 :         std::string grpName;
      47           1 :         const int32_t pid = mmGetPid();
      48           2 :         const std::lock_guard<std::mutex> groupLock(muForCreateGroup_);
      49           1 :         if (isGroupCreate) {
      50           0 :             return ACL_SUCCESS;
      51             :         }
      52           1 :         ACL_REQUIRES_OK(QueryGroup(pid, grpNum, grpName));
      53           1 :         if (grpNum == 0U) {
      54           1 :             ACL_LOG_INFO("need to create group");
      55           1 :             const rtMemGrpConfig_t grpConfig = {};
      56           2 :             const std::string gName = "acltdt" + std::to_string(pid);
      57           1 :             ACL_REQUIRES_CALL_RTS_OK(rtMemGrpCreate(gName.c_str(), &grpConfig), rtMemGrpCreate);
      58             : 
      59           1 :             rtMemGrpShareAttr_t shareAttr = {};
      60           1 :             shareAttr.admin = 1;
      61           1 :             shareAttr.read = 1;
      62           1 :             shareAttr.write = 1;
      63           1 :             shareAttr.alloc = 1;
      64           1 :             ACL_REQUIRES_CALL_RTS_OK(rtMemGrpAddProc(gName.c_str(), pid, &shareAttr), rtMemGrpAddProc);
      65           1 :             ACL_REQUIRES_CALL_RTS_OK(rtMemGrpAttach(gName.c_str(), 0), rtMemGrpAttach);
      66           1 :             ACL_REQUIRES_OK(QueryGroupId(gName));
      67             :         } else {
      68           0 :             ACL_REQUIRES_OK(QueryGroupId(grpName));
      69             :         }
      70           1 :         ACL_REQUIRES_OK(MbufInit());
      71           1 :         isGroupCreate = true;
      72           1 :         return ACL_SUCCESS;
      73             :     }
      74             : 
      75           1 :     aclError QueueProcessorCcpu::acltdtDestroyQueue(const uint32_t qid)
      76             :     {
      77           1 :         return acltdtDestroyQueueOndevice(qid, true);
      78             :     }
      79             : 
      80           1 :     aclError QueueProcessorCcpu::acltdtBindQueueRoutes(acltdtQueueRouteList *const qRouteList)
      81             :     {
      82           1 :         ACL_REQUIRES_NOT_NULL(qRouteList);
      83           1 :         ACL_LOG_INFO("Start to acltdtBindQueueRoutes, queue route is %zu", qRouteList->routeList.size());
      84             :         // qs is thread mode, so no need to grant queue to qs
      85           1 :         constexpr int32_t deviceId = 0;
      86           1 :         ACL_REQUIRES_OK(InitQueueSchedule(deviceId));
      87             :         // get dst id
      88           1 :         const int32_t dstPid = mmGetPid();
      89           1 :         rtEschedEventSummary_t eventSum = {0, 0U, 0, 0U, 0U, nullptr, 0U, 0};
      90           1 :         rtEschedEventReply_t ack = {nullptr, 0U, 0U};
      91           1 :         bqs::QsProcMsgRsp qsRsp = {0UL, 0, 0U, 0U, 0U, {0}};
      92           1 :         eventSum.pid = dstPid;
      93           1 :         eventSum.grpId = bqs::BIND_QUEUE_GROUP_ID;
      94           1 :         eventSum.eventId = RT_MQ_SCHED_EVENT_QS_MSG;
      95           1 :         eventSum.dstEngine = static_cast<uint32_t>(RT_MQ_DST_ENGINE_CCPU_DEVICE);
      96           1 :         ack.buf = reinterpret_cast<char_t *>(&qsRsp);
      97           1 :         ack.bufLen = sizeof(qsRsp);
      98           2 :         const std::lock_guard<std::recursive_mutex> lk(muForQueueCtrl_);
      99           1 :         if (!isQsInit_) {
     100           1 :             ACL_REQUIRES_OK(SendConnectQsMsg(deviceId, eventSum, ack));
     101           1 :             ACL_REQUIRES_CALL_RTS_OK(rtMemQueueAttach(deviceId, qsContactId_, 0), rtMemQueueAttach);
     102           1 :             isQsInit_ = true;
     103             :         }
     104           1 :         ACL_REQUIRES_OK(SendBindUnbindMsgOnDevice(qRouteList, true, eventSum, ack));
     105           1 :         ACL_LOG_INFO("Successfully to execute acltdtBindQueueRoutes, queue route is %zu",
     106             :                      qRouteList->routeList.size());
     107           1 :         return ACL_SUCCESS;
     108             :     }
     109             : 
     110           1 :     aclError QueueProcessorCcpu::acltdtUnbindQueueRoutes(acltdtQueueRouteList *const qRouteList)
     111             :     {
     112           1 :         ACL_REQUIRES_NOT_NULL(qRouteList);
     113           1 :         ACL_LOG_INFO("Start to acltdtUnBindQueueRoutes, queue route is %zu", qRouteList->routeList.size());
     114             :         // get dst id
     115           1 :         const int32_t dstPid = mmGetPid();
     116           1 :         rtEschedEventSummary_t eventSum = {0, 0U, 0, 0U, 0U, nullptr, 0U, 0};
     117           1 :         rtEschedEventReply_t ack = {nullptr, 0U, 0U};
     118           1 :         bqs::QsProcMsgRsp qsRsp = {0UL, 0, 0U, 0U, 0U, {0}};
     119           1 :         eventSum.pid = dstPid;
     120           1 :         eventSum.grpId = bqs::BIND_QUEUE_GROUP_ID;
     121           1 :         eventSum.eventId = RT_MQ_SCHED_EVENT_QS_MSG;
     122           1 :         eventSum.dstEngine = static_cast<uint32_t>(RT_MQ_DST_ENGINE_CCPU_DEVICE);
     123           1 :         ack.buf = reinterpret_cast<char_t *>(&qsRsp);
     124           1 :         ack.bufLen = sizeof(qsRsp);
     125           2 :         const std::lock_guard<std::recursive_mutex> lk(muForQueueCtrl_);
     126           1 :         ACL_REQUIRES_OK(SendBindUnbindMsgOnDevice(qRouteList, false, eventSum, ack));
     127           1 :         ACL_LOG_INFO("Successfully to execute acltdtUnBindQueueRoutes, queue route is %zu",
     128             :                      qRouteList->routeList.size());
     129           1 :         return ACL_SUCCESS;
     130             :     }
     131             : 
     132           1 :     aclError QueueProcessorCcpu::acltdtQueryQueueRoutes(const acltdtQueueRouteQueryInfo *const queryInfo,
     133             :                                                         acltdtQueueRouteList *const qRouteList)
     134             :     {
     135           1 :         ACL_REQUIRES_NOT_NULL(queryInfo);
     136           1 :         ACL_REQUIRES_NOT_NULL(qRouteList);
     137           1 :         ACL_LOG_INFO("Start to acltdtQueryQueueRoutes");
     138           1 :         constexpr int32_t deviceId = 0;
     139             :         // get dst id
     140           1 :         const int32_t dstPid = mmGetPid();
     141           1 :         rtEschedEventSummary_t eventSum = {0, 0U, 0, 0U, 0U, nullptr, 0U, 0};
     142           1 :         rtEschedEventReply_t ack = {nullptr, 0U, 0U};
     143           1 :         bqs::QsProcMsgRsp qsRsp = {0UL, 0, 0U, 0U, 0U, {0}};
     144           1 :         eventSum.pid = dstPid;
     145           1 :         eventSum.grpId = bqs::BIND_QUEUE_GROUP_ID;
     146           1 :         eventSum.eventId = RT_MQ_SCHED_EVENT_QS_MSG;
     147           1 :         eventSum.dstEngine = static_cast<uint32_t>(RT_MQ_DST_ENGINE_CCPU_DEVICE);
     148           1 :         ack.buf = reinterpret_cast<char_t *>(&qsRsp);
     149           1 :         ack.bufLen = sizeof(qsRsp);
     150           2 :         const std::lock_guard<std::recursive_mutex> lk(muForQueueCtrl_);
     151           1 :         size_t routeNum = 0UL;
     152           1 :         ACL_REQUIRES_OK(GetQueueRouteNum(queryInfo, deviceId, eventSum, ack, routeNum));
     153           1 :         ACL_REQUIRES_OK(QueryQueueRoutesOnDevice(queryInfo, routeNum, eventSum, ack, qRouteList));
     154           1 :         return ACL_SUCCESS;
     155             :     }
     156             : 
     157           1 :     aclError QueueProcessorCcpu::QueryGroup(const int32_t pid, size_t &grpNum, std::string &grpName) const
     158             :     {
     159           1 :         rtMemGrpQueryInput_t input = {};
     160           1 :         input.cmd = RT_MEM_GRP_QUERY_GROUPS_OF_PROCESS;
     161           1 :         input.grpQueryByProc.pid = pid;
     162           1 :         rtMemGrpQueryOutput_t output = {};
     163           1 :         rtMemGrpOfProc_t outputInfo[QUERY_BUFF_GRP_MAX_NUM] = {{}};
     164           1 :         output.groupsOfProc = outputInfo;
     165           1 :         output.maxNum = QUERY_BUFF_GRP_MAX_NUM;
     166             : 
     167           1 :         ACL_REQUIRES_CALL_RTS_OK(rtMemGrpQuery(&input, &output), rtMemGrpQuery);
     168           1 :         grpNum = output.resultNum;
     169           1 :         if (grpNum > 0) {
     170           0 :             grpName = std::string(output.groupsOfProc->groupName);
     171             :         }
     172           1 :         ACL_LOG_INFO("This proc [%d] has [%zu] group, name is %s", input.grpQueryByProc.pid,
     173             :                      grpNum, grpName.c_str());
     174             : 
     175           1 :         return ACL_SUCCESS;
     176             :     }
     177             : 
     178           1 :     aclError QueueProcessorCcpu::MbufInit() const
     179             :     {
     180             :         static bool isMbufInit = false;
     181           1 :         if (!isMbufInit) {
     182           1 :             rtMemBuffCfg_t cfg = {{}};
     183           1 :             const rtError_t ret = rtMbufInit(&cfg);
     184           1 :             if ((ret != ACL_RT_SUCCESS) && (ret != ACL_ERROR_RT_REPEATED_INIT)) {
     185           0 :                 ACL_LOG_INNER_ERROR("mbuf init failed, ret is %d", ret);
     186           0 :                 return ret;
     187             :             }
     188           1 :             isMbufInit = true;
     189             :         }
     190           1 :         return ACL_SUCCESS;
     191             :     }
     192             : 
     193           1 :     aclError QueueProcessorCcpu::acltdtAllocBuf(const size_t size, const uint32_t type, acltdtBuf *const buf)
     194             :     {
     195           1 :         ACL_REQUIRES_OK(acltdtCreateGroup());
     196           1 :         ACL_REQUIRES_OK(acltdtAllocBufData(size, type, buf));
     197           1 :         return ACL_SUCCESS;
     198             :     }
     199             : }