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              : 
      11              : #include "operator_kernel_model_report_status.h"
      12              : 
      13              : #include "dynamic_sched.pb.h"
      14              : #include "aicpusd_status.h"
      15              : #include "aicpusd_drv_manager.h"
      16              : #include "aicpusd_model_execute.h"
      17              : 
      18              : namespace AicpuSchedule {
      19              : namespace {
      20              : const std::string KERNEL_MODEL_REPORT_STATUS = "modelReportStatus";
      21              : } // namespace
      22              : 
      23            9 : int32_t OperatorKernelModelReportStatus::Compute(const AicpuTaskInfo& kernelTaskInfo, const RunContext& taskContext)
      24              : {
      25              :     const ReportStatusInfo* const bufInfo =
      26            9 :         PtrToPtr<void, ReportStatusInfo>(ValueToPtr(static_cast<uintptr_t>(kernelTaskInfo.paraBase)));
      27            9 :     if (bufInfo == nullptr) {
      28            1 :         aicpusd_err(
      29              :             "ModelReportStatus kernelTaskInfo paramBase is null, modelId[%u], streamId[%u], taskId[%u]",
      30              :             taskContext.modelId, taskContext.streamId, kernelTaskInfo.taskID);
      31            1 :         return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
      32              :     }
      33            8 :     std::vector<QueueAttrs> inputQueues;
      34              :     // 1 means queue info memory is after ReportStatusInfo memory
      35            8 :     const QueueAttrs* queuePtr = reinterpret_cast<const QueueAttrs*>(bufInfo + 1);
      36           16 :     for (uint32_t idx = 0U; idx < bufInfo->inputNum; idx++) {
      37            8 :         inputQueues.emplace_back(*queuePtr);
      38            8 :         queuePtr++;
      39              :     }
      40            8 :     const auto ret = ModelReportStatus(bufInfo->modelUuid, bufInfo->statusOutputQueue, inputQueues, taskContext);
      41            8 :     return ret;
      42            8 : }
      43              : 
      44            8 : int32_t OperatorKernelModelReportStatus::ModelReportStatus(
      45              :     const uint32_t modelUuid, const QueueAttrs& schedOutputQueue, const std::vector<QueueAttrs>& inputQueues,
      46              :     const RunContext& taskContext) const
      47              : {
      48            8 :     AicpuModel* const model = AicpuModelManager::GetInstance().GetModel(taskContext.modelId);
      49            8 :     if (model == nullptr) {
      50            1 :         aicpusd_err("cannot get model by modelId:[%u]!", taskContext.modelId);
      51            1 :         return AICPU_SCHEDULE_ERROR_INNER_ERROR;
      52              :     }
      53            7 :     uint32_t& inputConsumeNum = model->GetInputConsumeNumRef();
      54            7 :     inputConsumeNum++;
      55              :     // construct SubmodelStatus protobuf object
      56            7 :     const auto deviceId = AicpuDrvManager::GetInstance().GetDeviceId();
      57            7 :     aicpu::dynamci_sched::SubmodelStatus submodelStatus;
      58            7 :     submodelStatus.set_model_uuid(modelUuid);
      59           14 :     for (const auto& inputQueue : inputQueues) {
      60            7 :         const uint32_t inputQueueId = inputQueue.queueId;
      61            7 :         uint32_t queueDepth = UINT32_MAX;
      62              :         QueueInfo queueInfo;
      63            7 :         const auto drvRet = halQueueQueryInfo(deviceId, inputQueueId, &queueInfo);
      64            7 :         if (drvRet != DRV_ERROR_NONE) {
      65            1 :             aicpusd_info(
      66              :                 "Querying queue info was not successful, queue id[%u], device id[%u], ret[%d].", inputQueueId, deviceId,
      67              :                 drvRet);
      68              :         } else {
      69            6 :             queueDepth = static_cast<size_t>(queueInfo.size);
      70              :         }
      71            7 :         auto queueStatus = submodelStatus.add_queue_statuses();
      72            7 :         queueStatus->set_queue_depth(queueDepth);
      73            7 :         queueStatus->set_input_consume_num(inputConsumeNum);
      74            7 :         auto queueAttrs = queueStatus->mutable_queue_attrs();
      75            7 :         queueAttrs->set_queue_id(inputQueueId);
      76            7 :         queueAttrs->set_device_type(inputQueue.deviceType);
      77            7 :         queueAttrs->set_device_id(inputQueue.deviceId);
      78            7 :         queueAttrs->set_logic_id(inputQueue.logicId);
      79              :     }
      80              :     // enqueue
      81            7 :     const size_t reqSize = submodelStatus.ByteSizeLong();
      82           18 :     const FillFunc fillFunc = [&submodelStatus](void* const buffer, const size_t size) {
      83            4 :         if (submodelStatus.SerializeToArray(buffer, static_cast<int32_t>(size))) {
      84            4 :             return AICPU_SCHEDULE_OK;
      85              :         }
      86            0 :         aicpusd_err("Protobuf serializeToArray failed.");
      87            0 :         return AICPU_SCHEDULE_ERROR_INNER_ERROR;
      88            7 :     };
      89            7 :     const int32_t ret = EnqueueStatus(deviceId, schedOutputQueue.queueId, reqSize, fillFunc);
      90            7 :     if (ret == AICPU_SCHEDULE_OK) {
      91            1 :         inputConsumeNum = 0U;
      92            6 :     } else if (ret != AICPU_SCHEDULE_ERROR_QUEUE_FULL) {
      93            5 :         aicpusd_err("enqueue failed, deviceId[%u], queueId[%u], ret[%d].", deviceId, schedOutputQueue.queueId, ret);
      94            5 :         return ret;
      95              :     }
      96            2 :     aicpusd_info(
      97              :         "Dynamic sched report status, ret[%d], status[%s], deviceId[%u], queueId[%u]", ret,
      98              :         submodelStatus.DebugString().c_str(), deviceId, schedOutputQueue.queueId);
      99            2 :     return AICPU_SCHEDULE_OK;
     100            7 : }
     101              : 
     102            7 : int32_t OperatorKernelModelReportStatus::EnqueueStatus(
     103              :     const uint32_t deviceId, const uint32_t queueId, const size_t reqSize, const FillFunc& fillFunc) const
     104              : {
     105              :     // alloc mbuf
     106            7 :     Mbuf* mbuf = nullptr;
     107            7 :     auto drvRet = halMbufAlloc(reqSize, &mbuf);
     108            7 :     if (drvRet != DRV_ERROR_NONE) {
     109            1 :         aicpusd_err("halMbufAlloc failed, drvRet=%d, dataSize=%lu.", drvRet, reqSize);
     110            1 :         return AICPU_SCHEDULE_ERROR_DRV_ERR;
     111              :     }
     112            0 :     auto mbufDeleter = [](Mbuf* buf) { (void)halMbufFree(buf); };
     113            6 :     std::unique_ptr<Mbuf, decltype(mbufDeleter)> mbufGuard(mbuf, mbufDeleter);
     114            6 :     drvRet = halMbufSetDataLen(mbuf, reqSize);
     115            6 :     if (drvRet != DRV_ERROR_NONE) {
     116            1 :         aicpusd_err("halMbufSetDataLen failed, drvRet=%d, dataSize=%lu.", drvRet, reqSize);
     117            1 :         return AICPU_SCHEDULE_ERROR_DRV_ERR;
     118              :     }
     119              :     // write mbuf data
     120            5 :     void* buffAddr = nullptr;
     121            5 :     drvRet = halMbufGetBuffAddr(mbuf, &buffAddr);
     122            5 :     if (drvRet != DRV_ERROR_NONE || buffAddr == nullptr) {
     123            1 :         aicpusd_err("Failed to get buff addr, ret[%d].", drvRet);
     124            1 :         return AICPU_SCHEDULE_ERROR_DRV_ERR;
     125              :     }
     126            4 :     const auto ret = fillFunc(buffAddr, reqSize);
     127            4 :     if (ret != AICPU_SCHEDULE_OK) {
     128            0 :         aicpusd_err("Failed to fill mbuf data, ret[%d].", ret);
     129            0 :         return ret;
     130              :     }
     131              :     // enqueue
     132            4 :     drvRet = halQueueEnQueue(deviceId, queueId, mbuf);
     133            4 :     if (drvRet == DRV_ERROR_QUEUE_FULL) {
     134            1 :         aicpusd_debug("Queue[%u] is full.", queueId);
     135            1 :         return AICPU_SCHEDULE_ERROR_QUEUE_FULL;
     136            3 :     } else if (drvRet != DRV_ERROR_NONE) {
     137            2 :         aicpusd_err("Failed to enqueue, queueId[%u], ret[%d].", queueId, drvRet);
     138            2 :         return AICPU_SCHEDULE_ERROR_DRV_ERR;
     139              :     }
     140            1 :     (void)mbufGuard.release();
     141            1 :     return AICPU_SCHEDULE_OK;
     142            6 : }
     143              : 
     144            6 : REGISTER_OPERATOR_KERNEL(KERNEL_MODEL_REPORT_STATUS, OperatorKernelModelReportStatus);
     145              : } // namespace AicpuSchedule