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              : 
      19              : namespace AicpuSchedule {
      20              : namespace {
      21              : const std::string KERNEL_MODEL_REPORT_STATUS = "modelReportStatus";
      22              : }  // namespace
      23              : 
      24            9 : int32_t OperatorKernelModelReportStatus::Compute(const AicpuTaskInfo &kernelTaskInfo, const RunContext &taskContext)
      25              : {
      26              :     const ReportStatusInfo * const bufInfo =
      27            9 :         PtrToPtr<void, ReportStatusInfo>(ValueToPtr(static_cast<uintptr_t>(kernelTaskInfo.paraBase)));
      28            9 :     if (bufInfo == nullptr) {
      29            1 :         aicpusd_err("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(const uint32_t modelUuid, const QueueAttrs &schedOutputQueue,
      45              :                                                            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("Querying queue info was not successful, queue id[%u], device id[%u], ret[%d].",
      66              :                 inputQueueId, deviceId, drvRet);
      67              :         } else {
      68            6 :             queueDepth = static_cast<size_t>(queueInfo.size);
      69              :         }
      70            7 :         auto queueStatus = submodelStatus.add_queue_statuses();
      71            7 :         queueStatus->set_queue_depth(queueDepth);
      72            7 :         queueStatus->set_input_consume_num(inputConsumeNum);
      73            7 :         auto queueAttrs = queueStatus->mutable_queue_attrs();
      74            7 :         queueAttrs->set_queue_id(inputQueueId);
      75            7 :         queueAttrs->set_device_type(inputQueue.deviceType);
      76            7 :         queueAttrs->set_device_id(inputQueue.deviceId);
      77            7 :         queueAttrs->set_logic_id(inputQueue.logicId);
      78              :     }
      79              :     // enqueue
      80            7 :     const size_t reqSize = submodelStatus.ByteSizeLong();
      81           18 :     const FillFunc fillFunc = [&submodelStatus](void *const buffer, const size_t size) {
      82            4 :         if (submodelStatus.SerializeToArray(buffer, static_cast<int32_t>(size))) {
      83            4 :             return AICPU_SCHEDULE_OK;
      84              :         }
      85            0 :         aicpusd_err("Protobuf serializeToArray failed.");
      86            0 :         return AICPU_SCHEDULE_ERROR_INNER_ERROR;
      87            7 :     };
      88            7 :     const int32_t ret = EnqueueStatus(deviceId, schedOutputQueue.queueId, reqSize, fillFunc);
      89            7 :     if (ret == AICPU_SCHEDULE_OK) {
      90            1 :         inputConsumeNum = 0U;
      91            6 :     } else if (ret != AICPU_SCHEDULE_ERROR_QUEUE_FULL) {
      92            5 :         aicpusd_err("enqueue failed, deviceId[%u], queueId[%u], ret[%d].",
      93              :             deviceId, schedOutputQueue.queueId, ret);
      94            5 :         return ret;
      95              :     }
      96            2 :     aicpusd_info("Dynamic sched report status, ret[%d], status[%s], deviceId[%u], queueId[%u]",
      97              :         ret, submodelStatus.DebugString().c_str(), deviceId, schedOutputQueue.queueId);
      98            2 :     return AICPU_SCHEDULE_OK;
      99            7 : }
     100              : 
     101            7 : int32_t OperatorKernelModelReportStatus::EnqueueStatus(const uint32_t deviceId, const uint32_t queueId,
     102              :                                                        const size_t reqSize, const FillFunc &fillFunc) const
     103              : {
     104              :     // alloc mbuf
     105            7 :     Mbuf *mbuf = nullptr;
     106            7 :     auto drvRet = halMbufAlloc(reqSize, &mbuf);
     107            7 :     if (drvRet != DRV_ERROR_NONE) {
     108            1 :         aicpusd_err("halMbufAlloc failed, drvRet=%d, dataSize=%lu.", drvRet, reqSize);
     109            1 :         return AICPU_SCHEDULE_ERROR_DRV_ERR;
     110              :     }
     111            0 :     auto mbufDeleter = [](Mbuf *buf) { (void)halMbufFree(buf); };
     112            6 :     std::unique_ptr<Mbuf, decltype(mbufDeleter)> mbufGuard(mbuf, mbufDeleter);
     113            6 :     drvRet = halMbufSetDataLen(mbuf, reqSize);
     114            6 :     if (drvRet != DRV_ERROR_NONE) {
     115            1 :         aicpusd_err("halMbufSetDataLen failed, drvRet=%d, dataSize=%lu.", drvRet, reqSize);
     116            1 :         return AICPU_SCHEDULE_ERROR_DRV_ERR;
     117              :     }
     118              :     // write mbuf data
     119            5 :     void *buffAddr = nullptr;
     120            5 :     drvRet = halMbufGetBuffAddr(mbuf, &buffAddr);
     121            5 :     if (drvRet != DRV_ERROR_NONE || buffAddr == nullptr) {
     122            1 :         aicpusd_err("Failed to get buff addr, ret[%d].", drvRet);
     123            1 :         return AICPU_SCHEDULE_ERROR_DRV_ERR;
     124              :     }
     125            4 :     const auto ret = fillFunc(buffAddr, reqSize);
     126            4 :     if (ret != AICPU_SCHEDULE_OK) {
     127            0 :         aicpusd_err("Failed to fill mbuf data, ret[%d].", ret);
     128            0 :         return ret;
     129              :     }
     130              :     // enqueue
     131            4 :     drvRet = halQueueEnQueue(deviceId, queueId, mbuf);
     132            4 :     if (drvRet == DRV_ERROR_QUEUE_FULL) {
     133            1 :         aicpusd_debug("Queue[%u] is full.", queueId);
     134            1 :         return AICPU_SCHEDULE_ERROR_QUEUE_FULL;
     135            3 :     } else if (drvRet != DRV_ERROR_NONE) {
     136            2 :         aicpusd_err("Failed to enqueue, queueId[%u], ret[%d].", queueId, drvRet);
     137            2 :         return AICPU_SCHEDULE_ERROR_DRV_ERR;
     138              :     }
     139            1 :     (void)mbufGuard.release();
     140            1 :     return AICPU_SCHEDULE_OK;
     141            6 : }
     142              : 
     143              : 
     144            6 : REGISTER_OPERATOR_KERNEL(KERNEL_MODEL_REPORT_STATUS, OperatorKernelModelReportStatus);
     145              : }  // namespace AicpuSchedule