/**
 * Copyright (c) 2025 Huawei Technologies Co., Ltd.
 * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
 * CANN Open Software License Agreement Version 2.0 (the "License").
 * Please refer to the License for details. You may not use this file except in compliance with the License.
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
 * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
 * See LICENSE in the root of the software repository for the full text of the License.
 */

#include "tensor_data_transfer.h"
#include <map>
#include <mutex>
#include <sstream>
#include <unordered_map>

#include "data_common.h"
#include "tdt_host_interface.h"

#include "log_inner.h"
#include "acl/acl_tdt_queue.h"
#include "acl_tdt_queue/queue.h"
#include "runtime/rt_mem_queue.h"
#include "runtime/mem.h"
#include "runtime/context.h"
#include "runtime/rts/rts_mem.h"
#include "utils/file_utils.h"
#include "utils/data_type_utils.h"

namespace {
std::mutex aclChannleMutex;
std::map<std::string, acltdtChannelHandle*> aclChannleMap;
std::map<std::string, aclDataType> aclDataTypeStrMap = {
    {"bool", ACL_BOOL},     {"int8", ACL_INT8},     {"uint8", ACL_UINT8},  {"half", ACL_FLOAT16},  {"int16", ACL_INT16},
    {"uint16", ACL_UINT16}, {"float", ACL_FLOAT},   {"int32", ACL_INT32},  {"uint32", ACL_UINT32}, {"int64", ACL_INT64},
    {"uint64", ACL_UINT64}, {"double", ACL_DOUBLE}, {"string", ACL_STRING}};
constexpr uint32_t VERSION_NAME = 1U;
constexpr size_t TDT_TENSOR_ALIGNE_UNIT = 64UL;
const std::vector<size_t> GEAR_SIZE{
    1U * 1024U * 1024U, 10U * 1024U * 1024U, 100U * 1024U * 1024U, 500U * 1024U * 1024U};
size_t Get64AlignedSize(const size_t size)
{
    return (size + TDT_TENSOR_ALIGNE_UNIT - 1UL) / TDT_TENSOR_ALIGNE_UNIT * TDT_TENSOR_ALIGNE_UNIT;
}

using TdtHostInitFunc = int32_t (*)(uint32_t);
using TdtHostPreparePopDataFunc = int32_t (*)();
using TdtHostPopDataFunc = int32_t (*)(const std::string&, std::vector<tdt::DataItem>&);
using TdtHostPushDataFunc = int32_t (*)(const std::string&, const std::vector<tdt::DataItem>&, uint32_t deviceId);
using TdtHostStopFunc = int32_t (*)(const std::string&);
using TdtHostDestroyFunc = int32_t (*)();

#ifndef RUN_TEST
void* GetHandler()
{
    std::string soPath;
    if (acl::file_utils::GetSoRealPath(soPath) != ACL_SUCCESS) {
        ACL_LOG_ERROR("Get libacl_tdt_channel.so path failed.");
        return nullptr;
    }
    std::string soName = soPath + "libdatatransfer.so";
    // Load the "libdatatransfer.so" library until the program ends. During the process, Dlclose is not invoked
    // to prevent the destruction of the global state information saved in ibdatatransfer.so.
    void* handler = mmDlopen(soName.c_str(), RTLD_NOW | RTLD_GLOBAL);
    if (handler == nullptr) {
        ACL_LOG_ERROR(
            "The corresponding dependent dynamic library cannot be found. "
            "Please confirm whether the environment supports it and if the extension package has been correctly "
            "installed. "
            "soName is %s.",
            soName.c_str());
    }
    return handler;
}
#endif

void* GetFunction(const std::string& func_name)
{
#ifdef RUN_TEST
    std::unordered_map<std::string, void*> stubFunctionMap = {
        {"TdtHostInit", reinterpret_cast<void*>(&tdt::TdtHostInit)},
        {"TdtHostPushData", reinterpret_cast<void*>(&tdt::TdtHostPushData)},
        {"TdtHostDestroy", reinterpret_cast<void*>(&tdt::TdtHostDestroy)},
        {"TdtHostPreparePopData", reinterpret_cast<void*>(&tdt::TdtHostPreparePopData)},
        {"TdtHostPopData", reinterpret_cast<void*>(&tdt::TdtHostPopData)},
        {"TdtHostStop", reinterpret_cast<void*>(&tdt::TdtHostStop)}};
    auto it = stubFunctionMap.find(func_name);
    if (it != stubFunctionMap.end()) {
        return it->second;
    }
    return nullptr;
#else
    static void* handler = GetHandler();
    if (handler == nullptr) {
        ACL_LOG_ERROR("Get handler failed when get %s function.", func_name.c_str());
        return nullptr;
    }
    void* func_ptr = mmDlsym(handler, func_name.c_str());
    if (func_ptr == nullptr) {
        ACL_LOG_ERROR(
            "The corresponding symbol cannot be found. Please confirm whether the installed extension package is "
            "correct, %s.",
            mmDlerror());
    }
    return func_ptr;
#endif
}
} // namespace

namespace acl {
bool GetTensorShape(const std::string& dimsStr, std::vector<int64_t>& dims)
{
    // change "[32,224,224,3]" => "32,224,224,3"
    // tensor_shape.size() - 2 is the second to last
    if (dimsStr.size() < 2) {
        ACL_LOG_INNER_ERROR("[Check][dimsStr]Invalid shape string: %s", dimsStr.c_str());
        return false;
    }

    std::string str = dimsStr.substr(1, dimsStr.size() - 2);
    if (!str.empty()) {
        std::string::size_type index = 0;
        while ((index = str.find(' ', index)) != std::string::npos) {
            (void)str.erase(index, 1);
        }
    }
    std::string split = ",";
    std::string::size_type pos2 = str.find(split);
    std::string::size_type pos1 = 0;
    while (pos2 != std::string::npos) {
        try {
            dims.push_back(std::stoll(str.substr(pos1, pos2 - pos1)));
        } catch (...) {
            ACL_LOG_INNER_ERROR("[Check][Shape]Invalid shape string: %s", dimsStr.c_str());
            return false;
        }
        // string::size_type can store the length of any string object
        pos1 = pos2 + split.size();
        pos2 = str.find(split, pos1);
    }
    if (pos1 != str.length()) {
        try {
            dims.push_back(std::stoll(str.substr(pos1)));
        } catch (...) {
            ACL_LOG_INNER_ERROR("[Check][Shape]Invalid shape string: %s", dimsStr.c_str());
            return false;
        }
    }
    return true;
}

aclError GetTdtDataTypeByAclDataType(acltdtTensorType aclType, tdt::TdtDataType& tdtDataType)
{
    switch (aclType) {
        case ACL_TENSOR_DATA_END_OF_SEQUENCE: {
            tdtDataType = tdt::TDT_END_OF_SEQUENCE;
            break;
        }
        case ACL_TENSOR_DATA_TENSOR: {
            tdtDataType = tdt::TDT_TENSOR;
            break;
        }
        case ACL_TENSOR_DATA_ABNORMAL: {
            tdtDataType = tdt::TDT_ABNORMAL;
            break;
        }
        default: {
            acl::AclErrorLogManager::ReportInputError(
                acl::INVALID_VALUE_MSG, std::vector<const char*>({"func", "value", "param", "expect"}),
                std::vector<const char*>(
                    {"Obtaining the tdt data type", acl::GetTensorTypeDesc(aclType), "aclType",
                     "ACL_TENSOR_DATA_END_OF_SEQUENCE or ACL_TENSOR_DATA_TENSOR or ACL_TENSOR_DATA_ABNORMAL"}));
            return ACL_ERROR_INVALID_PARAM;
        }
    }
    return ACL_SUCCESS;
}

aclError GetTdtDataTypeByAclDataTypeV2(acltdtTensorType aclType, int32_t& tdtDataType)
{
    switch (aclType) {
        case ACL_TENSOR_DATA_END_OF_SEQUENCE: {
            tdtDataType = 1;
            break;
        }
        case ACL_TENSOR_DATA_TENSOR: {
            tdtDataType = 0;
            break;
        }
        case ACL_TENSOR_DATA_ABNORMAL: {
            tdtDataType = 2;
            break;
        }
        default: {
            acl::AclErrorLogManager::ReportInputError(
                acl::INVALID_VALUE_MSG, std::vector<const char*>({"func", "value", "param", "expect"}),
                std::vector<const char*>(
                    {"Obtaining the tdt data type", acl::GetTensorTypeDesc(aclType), "aclType",
                     "ACL_TENSOR_DATA_END_OF_SEQUENCE or ACL_TENSOR_DATA_TENSOR or ACL_TENSOR_DATA_ABNORMAL"}));
            return ACL_ERROR_INVALID_PARAM;
        }
    }
    return ACL_SUCCESS;
}

aclError GetAclTypeByTdtDataType(tdt::TdtDataType tdtDataType, acltdtTensorType& aclType)
{
    switch (tdtDataType) {
        case tdt::TDT_END_OF_SEQUENCE: {
            aclType = ACL_TENSOR_DATA_END_OF_SEQUENCE;
            break;
        }
        case tdt::TDT_TENSOR: {
            aclType = ACL_TENSOR_DATA_TENSOR;
            break;
        }
        case tdt::TDT_ABNORMAL: {
            aclType = ACL_TENSOR_DATA_ABNORMAL;
            break;
        }
        default: {
            acl::AclErrorLogManager::ReportInputError(
                acl::INVALID_VALUE_MSG, std::vector<const char*>({"func", "value", "param", "expect"}),
                std::vector<const char*>(
                    {"Obtaining the tensor data type", acl::GetTdtDataTypeDesc(tdtDataType), "tdtDataType",
                     "TDT_END_OF_SEQUENCE or TDT_TENSOR or TDT_ABNORMAL"}));
            return ACL_ERROR_UNSUPPORTED_DATA_TYPE;
        }
    }
    return ACL_SUCCESS;
}

aclError GetAclTypeByTdtDataTypeV2(int32_t tdtDataType, acltdtTensorType& aclType)
{
    switch (tdtDataType) {
        case 1: {
            aclType = ACL_TENSOR_DATA_END_OF_SEQUENCE;
            break;
        }
        case 0: {
            aclType = ACL_TENSOR_DATA_TENSOR;
            break;
        }
        case 2: {
            aclType = ACL_TENSOR_DATA_ABNORMAL;
            break;
        }
        case 3: {
            aclType = ACL_TENSOR_DATA_SLICE_TENSOR;
            break;
        }
        case 4: {
            aclType = ACL_TENSOR_DATA_END_TENSOR;
            break;
        }
        default: {
            acl::AclErrorLogManager::ReportInputError(
                acl::INVALID_VALUE_MSG, std::vector<const char*>({"func", "value", "param", "expect"}),
                std::vector<const char*>(
                    {"Obtaining the tensor data type", acl::GetTdtDataTypeDescV2(tdtDataType), "tdtDataType",
                     "[TDT_TENSOR(0), TDT_END_OF_SEQUENCE(1), TDT_ABNORMAL(2), TDT_SLICE_TENSOR(3), "
                     "TDT_END_TENSOR(4)]"}));
            return ACL_ERROR_UNSUPPORTED_DATA_TYPE;
        }
    }
    return ACL_SUCCESS;
}

aclError TensorDatasetSerializes(const acltdtDataset* dataset, std::vector<tdt::DataItem>& itemVec)
{
    ACL_REQUIRES_NOT_NULL(dataset);

    for (size_t i = 0; i < dataset->blobs.size(); ++i) {
        tdt::DataItem item;
        tdt::TdtDataType tdtDataType;
        const auto ret = GetTdtDataTypeByAclDataType(dataset->blobs[i]->tdtType, tdtDataType);
        if (ret != ACL_SUCCESS) {
            ACL_LOG_INNER_ERROR(
                "[Check][Dataset]TensorDatasetSerializes failed, "
                "invalid tdt type %d",
                dataset->blobs[i]->tdtType);
            itemVec.clear();
            return ret;
        }

        item.dataType_ = tdtDataType;
        item.tensorShape_ = dataset->blobs[i]->dimsStr;
        item.tensorType_ = dataset->blobs[i]->dataTypeStr;
        item.dataLen_ = dataset->blobs[i]->dataLen;
        item.dataPtr_ = dataset->blobs[i]->dataPtr;
        itemVec.emplace_back(item);
    }
    return ACL_SUCCESS;
}

aclError TensorDatasetSerializesV2(const acltdtDataset* dataset, std::vector<acl::aclTdtDataItemInfo>& itemVec)
{
    ACL_REQUIRES_NOT_NULL(dataset);
    for (size_t i = 0; i < dataset->blobs.size(); ++i) {
        acl::aclTdtDataItemInfo item;
        int32_t tdtDataType;
        const auto ret = GetTdtDataTypeByAclDataTypeV2(dataset->blobs[i]->tdtType, tdtDataType);
        if (ret != ACL_SUCCESS) {
            ACL_LOG_INNER_ERROR(
                "[Check][Dataset]TensorDatasetSerializes failed, "
                "invalid tdt type %d",
                dataset->blobs[i]->tdtType);
            return ret;
        }

        item.ctrlInfo.dataType = tdtDataType;
        item.ctrlInfo.tensorType = dataset->blobs[i]->dataType;
        item.ctrlInfo.dimNum = dataset->blobs[i]->dims.size();
        item.dims = dataset->blobs[i]->dims;
        item.ctrlInfo.dataLen = dataset->blobs[i]->dataLen;
        item.dataPtr = dataset->blobs[i]->dataPtr;
        itemVec.emplace_back(item);
        ACL_LOG_DEBUG(
            "TensorDatasetSerializesWithQueue, dataType %d, tensorType %d, dimNum %u, dataLen %lu",
            item.ctrlInfo.dataType, item.ctrlInfo.tensorType, item.ctrlInfo.dimNum, item.ctrlInfo.dataLen);
    }
    return ACL_SUCCESS;
}

aclError TensorDatasetDeserializes(const std::vector<tdt::DataItem>& itemVec, acltdtDataset* dataset)
{
    ACL_REQUIRES_NOT_NULL(dataset);
    if (dataset->blobs.size() != 0) {
        const std::string sizeVal = std::to_string(dataset->blobs.size());
        acl::AclErrorLogManager::ReportInputError(
            acl::INVALID_PARAM_REASON_MSG, std::vector<const char*>({"func", "value", "param", "reason"}),
            std::vector<const char*>(
                {"tdt data deserialization", sizeVal.c_str(), "dataset->blobs.size",
                 "dataset must be empty before deserialization"}));
        return ACL_ERROR_INVALID_PARAM;
    }
    aclError ret = ACL_SUCCESS;
    for (size_t i = 0; i < itemVec.size(); ++i) {
        acltdtTensorType aclType;
        ret = GetAclTypeByTdtDataType(itemVec[i].dataType_, aclType);
        if (ret != ACL_SUCCESS) {
            ACL_LOG_INNER_ERROR(
                "[Check][Dataset]TensorDatasetDeserializes failed, invalid data type %d", itemVec[i].dataType_);
            break;
        }

        if (aclType == ACL_TENSOR_DATA_TENSOR) {
            std::vector<int64_t> dims;
            if (!GetTensorShape(itemVec[i].tensorShape_, dims)) {
                ACL_LOG_INNER_ERROR(
                    "[Check][TensorDataset]TensorDatasetDeserializes failed, "
                    "invalid tensor shape[%s]",
                    itemVec[i].tensorShape_.c_str());
                ret = ACL_ERROR_INTERNAL_ERROR;
                break;
            }

            std::map<std::string, aclDataType>::const_iterator iter = aclDataTypeStrMap.find(itemVec[i].tensorType_);
            if (iter == aclDataTypeStrMap.cend()) {
                ACL_LOG_INNER_ERROR(
                    "[Deserialize][TensorDataset]TensorDatasetDeserializes failed, "
                    "unknown data type[%s]",
                    itemVec[i].tensorType_.c_str());
                ret = ACL_ERROR_INTERNAL_ERROR;
                break;
            }
            const aclDataType dataType = iter->second;
            acltdtDataItem* item = new (std::nothrow) acltdtDataItem(
                aclType, &dims[0], dims.size(), itemVec[i].tensorShape_, dataType, itemVec[i].tensorType_,
                itemVec[i].dataPtr_, itemVec[i].dataLen_);
            if (item == nullptr) {
                ACL_LOG_ERROR("[Check][Item]TensorDatasetDeserializes alloc failed");
                std::string sizeStr = std::to_string(sizeof(acltdtDataItem));
                acl::AclErrorLogManager::ReportInputError(
                    acl::ALLOC_MEMORY_FAILED_MSG, std::vector<const char*>({"buf_size", "alloc_interface"}),
                    std::vector<const char*>({sizeStr.c_str(), "new"}));
                ret = ACL_ERROR_BAD_ALLOC;
                break;
            }
            dataset->blobs.push_back(item);
        } else {
            acltdtDataItem* item = new (std::nothrow) acltdtDataItem(
                aclType, nullptr, 0, itemVec[i].tensorShape_, ACL_DT_UNDEFINED, itemVec[i].tensorType_,
                itemVec[i].dataPtr_, itemVec[i].dataLen_);
            if (item == nullptr) {
                ACL_LOG_INNER_ERROR("[Check][Item]TensorDatasetDeserializes alloc failed");
                ret = ACL_ERROR_BAD_ALLOC;
                break;
            }
            dataset->blobs.push_back(item);
        }
    }

    if (ret != ACL_SUCCESS) {
        for (size_t i = 0; i < dataset->blobs.size(); ++i) {
            ACL_DELETE_AND_SET_NULL(dataset->blobs[i]);
        }
        dataset->blobs.clear();
    }
    dataset->freeSelf = true;
    return ret;
}

aclError TensorDatasetDeserializesV2(const std::vector<acl::aclTdtDataItemInfo>& itemVec, acltdtDataset* dataset)
{
    ACL_REQUIRES_NOT_NULL(dataset);
    if (!dataset->blobs.empty() && !dataset->freeSelf) {
        const std::string sizeVal = std::to_string(dataset->blobs.size());
        acl::AclErrorLogManager::ReportInputError(
            acl::INVALID_PARAM_REASON_MSG, std::vector<const char*>({"func", "value", "param", "reason"}),
            std::vector<const char*>(
                {"tdt data deserialization", sizeVal.c_str(), "dataset->blobs.size",
                 "dataset must be empty or freeSelf must be true before deserialization"}));
        return ACL_ERROR_INVALID_PARAM;
    }
    for (auto it = dataset->blobs.begin(); it != dataset->blobs.end(); ++it) {
        ACL_DELETE_AND_SET_NULL(*it);
    }
    dataset->blobs.clear();
    aclError ret = ACL_SUCCESS;
    for (size_t i = 0; i < itemVec.size(); ++i) {
        acltdtTensorType aclType;
        ret = GetAclTypeByTdtDataTypeV2(itemVec[i].ctrlInfo.dataType, aclType);
        if (ret != ACL_SUCCESS) {
            ACL_LOG_INNER_ERROR(
                "[Check][Dataset]Failed to convert TdtDataType %d to acltdtTensorType.", itemVec[i].ctrlInfo.dataType);
            break;
        }
        if ((aclType == ACL_TENSOR_DATA_TENSOR) || (aclType == ACL_TENSOR_DATA_SLICE_TENSOR) ||
            (aclType == ACL_TENSOR_DATA_END_TENSOR)) {
            if (itemVec[i].ctrlInfo.version == static_cast<int32_t>(VERSION_NAME)) {
                void* dataReal =
                    (itemVec[i].priorityDataPtr_ != nullptr) ? itemVec[i].priorityDataPtr_ : itemVec[i].dataPtr.get();
                dataset->name.assign(static_cast<char*>(dataReal), itemVec[i].ctrlInfo.dataLen);
                ACL_LOG_INFO("get dataset name is %s", dataset->name.c_str());
                continue;
            }
            std::vector<int64_t> dims = itemVec[i].dims;
            const aclDataType dataType = static_cast<aclDataType>(itemVec[i].ctrlInfo.tensorType);
            acltdtDataItem* item = new (std::nothrow) acltdtDataItem(
                aclType, &dims[0], dims.size(), "", dataType, "", itemVec[i].dataPtr, itemVec[i].ctrlInfo.dataLen);
            if (item == nullptr) {
                ACL_LOG_INNER_ERROR("[Check][Item]TensorDatasetDeserializes alloc failed");
                ret = ACL_ERROR_BAD_ALLOC;
                break;
            }
            item->sliceNum = itemVec[i].ctrlInfo.sliceNum;
            item->sliceId = itemVec[i].ctrlInfo.sliceId;
            item->priorityData_ = itemVec[i].priorityDataPtr_;
            dataset->blobs.push_back(item);
        } else {
            acltdtDataItem* item = new (std::nothrow) acltdtDataItem(
                aclType, nullptr, 0, "", ACL_DT_UNDEFINED, "", itemVec[i].dataPtr, itemVec[i].ctrlInfo.dataLen);
            if (item == nullptr) {
                ACL_LOG_INNER_ERROR("[Check][Item]TensorDatasetDeserializes alloc failed");
                ret = ACL_ERROR_BAD_ALLOC;
                break;
            }
            item->priorityData_ = itemVec[i].priorityDataPtr_;
            dataset->blobs.push_back(item);
        }
    }

    if (ret != ACL_SUCCESS) {
        for (size_t i = 0; i < dataset->blobs.size(); ++i) {
            ACL_DELETE_AND_SET_NULL(dataset->blobs[i]);
        }
        dataset->blobs.clear();
    }
    dataset->freeSelf = true;
    return ret;
}

void GetTensorDimsString(const int64_t* dims, size_t dimNum, std::string& dimsStr)
{
    for (size_t i = 0; i < dimNum; ++i) {
        dimsStr += std::to_string(dims[i]);
        if (i + 1 == dimNum) {
            break;
        }
        dimsStr.push_back(',');
    }
    dimsStr += "]";
}

aclError SaveCtrlSharedPtrToVec(
    const datasetMemType memType, rtMemQueueBuffInfo& qItem, const std::shared_ptr<uint8_t>& ctrlSharedPtr,
    std::vector<std::shared_ptr<uint8_t>>& ctrlSharedPtrVec)
{
    void* ctrlPtr = ctrlSharedPtr.get();
    if (memType == MEM_DEVICE) {
        uint8_t* devPtr = nullptr;
        std::shared_ptr<uint8_t> ctrlSharedDevPtr;
        ctrlSharedDevPtr.reset(devPtr, [](void* p) {
            if (p != nullptr) {
                (void)rtFree(p);
            }
        });
        ACL_REQUIRES_RTS_OK_WARN_NOT_SUPPORT(
            rtMalloc(reinterpret_cast<void**>(&devPtr), qItem.len, RT_MEMORY_DEFAULT, acl::ACL_MODE_ID_U16), rtMalloc);
        ACL_REQUIRES_RTS_OK_WARN_NOT_SUPPORT(
            rtMemcpy(devPtr, qItem.len, ctrlPtr, qItem.len, RT_MEMCPY_HOST_TO_DEVICE), rtMemcpy);
        qItem.addr = devPtr;
        ctrlSharedPtrVec.push_back(ctrlSharedDevPtr);
    } else {
        qItem.addr = ctrlPtr;
        ctrlSharedPtrVec.push_back(ctrlSharedPtr);
    }
    return ACL_SUCCESS;
}

aclError UnpackageRecvDataInfo(uint8_t* outputHostAddr, size_t size, std::vector<acl::aclTdtDataItemInfo>& itemVec)
{
    ItemInfo* head = reinterpret_cast<ItemInfo*>(outputHostAddr);
    uint32_t cnt = head->cnt;
    ACL_LOG_INFO("get tensor cnt is %u", cnt);
    size_t offset = 0;
    for (uint32_t i = 0; i < cnt; ++i) {
        if (offset + sizeof(ItemInfo) > size) {
            ACL_LOG_ERROR("offset is %zu, size is %zu", offset, size);
            return ACL_ERROR_FAILURE;
        }
        acl::aclTdtDataItemInfo item;
        ItemInfo* tmp = reinterpret_cast<ItemInfo*>(outputHostAddr + offset);
        item.ctrlInfo = *tmp;
        ACL_LOG_INFO(
            "UnpackInfo version %d, dataType %d, curCnt %u, cnt %u, tensorType %d, dimNum %u, "
            "dynamicBitSize %u, sliceNum %u, sliceId %u, dataLen %lu",
            tmp->version, tmp->dataType, tmp->curCnt, tmp->cnt, tmp->tensorType, tmp->dimNum, tmp->dynamicBitSize,
            static_cast<uint32_t>(tmp->sliceNum), static_cast<uint32_t>(tmp->sliceId), tmp->dataLen);
        offset += sizeof(ItemInfo);

        for (uint32_t j = 0; j < tmp->dimNum; ++j) {
            if (offset + sizeof(int64_t) > size) {
                ACL_LOG_ERROR("offset is %zu, size is %zu", offset, size);
                return ACL_ERROR_FAILURE;
            }
            int64_t dimTmp = *(reinterpret_cast<int64_t*>(outputHostAddr + offset));
            item.dims.push_back(dimTmp);
            ACL_LOG_INFO("current dims[%u] is %ld", j, dimTmp);
            offset += sizeof(int64_t);
        }

        if (offset + tmp->dataLen > size) {
            ACL_LOG_ERROR("offset is %zu, data len is %lu, size is %zu", offset, tmp->dataLen, size);
            return ACL_ERROR_FAILURE;
        }
        if (tmp->dataLen > 0U) {
            item.priorityDataPtr_ = outputHostAddr + offset;
            offset += tmp->dataLen;
        } else {
            ACL_LOG_INFO("data length is 0");
        }
        ACL_LOG_INFO("after %u tensor, offset is %zu", i + 1, offset);
        itemVec.push_back(item);
    }
    return ACL_SUCCESS;
}

aclError TensorDataitemSerialize(
    std::vector<acl::aclTdtDataItemInfo>& itemVec, const datasetMemType memType,
    std::vector<rtMemQueueBuffInfo>& qBufVec, std::vector<std::shared_ptr<uint8_t>>& ctrlSharedPtrVec)
{
    uint32_t currentCnt = 0;
    size_t lastDataSize = 0U;
    for (size_t i = 0; i < itemVec.size(); ++i) {
        itemVec[i].ctrlInfo.curCnt = currentCnt;
        itemVec[i].ctrlInfo.cnt = itemVec.size();
        const size_t ctrlSize = sizeof(ItemInfo) + itemVec[i].dims.size() * sizeof(int64_t);
        // 64n + lastDataSize + 64n - lastDataSize
        const size_t alignedSize = Get64AlignedSize(ctrlSize + lastDataSize) - lastDataSize;
        itemVec[i].ctrlInfo.dynamicBitSize = alignedSize - sizeof(ItemInfo);
        std::shared_ptr<uint8_t> ctrlSharedPtr(
            new (std::nothrow) uint8_t[alignedSize], std::default_delete<uint8_t[]>());
        ACL_CHECK_MALLOC_RESULT_REPORT_RET(ctrlSharedPtr.get(), alignedSize, "new", ACL_ERROR_BAD_ALLOC);
        void* ctrlPtr = ctrlSharedPtr.get();
        ACL_LOG_DEBUG(
            "TensorDataitemSerialize alignedSize is %zu, ctrlSize is %zu, dynamicBitSize is %u, i is %zu,"
            " lastDataSize is %zu, shape size is %zu",
            alignedSize, ctrlSize, itemVec[i].ctrlInfo.dynamicBitSize, i, lastDataSize, itemVec[i].dims.size());
        auto memcpyRet = memcpy_s(ctrlPtr, alignedSize, &itemVec[i].ctrlInfo, sizeof(ItemInfo));
        if (memcpyRet != EN_OK) {
            ACL_LOG_INNER_ERROR(
                "[Call][MemCpy]call memcpy failed, result=%d, srcLen=%zu, dstLen=%zu", memcpyRet, sizeof(ItemInfo),
                alignedSize);
        }
        size_t offset = sizeof(ItemInfo);
        for (size_t j = 0; j < itemVec[i].dims.size(); ++j) {
            memcpyRet = memcpy_s(
                reinterpret_cast<uint8_t*>(ctrlPtr) + offset, alignedSize - offset, &itemVec[i].dims[j],
                sizeof(int64_t));
            if (memcpyRet != EN_OK) {
                ACL_LOG_INNER_ERROR(
                    "[Call][MemCpy]call memcpy failed, result=%d, srcLen=%zu, dstLen=%zu", memcpyRet, sizeof(int64_t),
                    alignedSize - offset);
            }
            offset += sizeof(int64_t);
        }
        rtMemQueueBuffInfo qItem = {};
        qItem.len = alignedSize;
        ACL_REQUIRES_OK(SaveCtrlSharedPtrToVec(memType, qItem, ctrlSharedPtr, ctrlSharedPtrVec));
        qBufVec.push_back(qItem);

        if (itemVec[i].ctrlInfo.dataLen > 0U) {
            rtMemQueueBuffInfo tmpQItem = {itemVec[i].dataPtr.get(), itemVec[i].ctrlInfo.dataLen};
            qBufVec.push_back(tmpQItem);
        } else {
            ACL_LOG_DEBUG("no need to insert data buf");
        }
        // current total size is (64n + lastDataSize)
        lastDataSize = itemVec[i].ctrlInfo.dataLen;
        ++currentCnt;
    }
    return ACL_SUCCESS;
}

aclError acltdtSendTensorV2(const acltdtChannelHandle* handle, const acltdtDataset* dataset, int32_t timeout)
{
    ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(handle);
    std::vector<acl::aclTdtDataItemInfo> itemVec;
    auto ret = acl::TensorDatasetSerializesV2(dataset, itemVec);
    if (ret != ACL_SUCCESS) {
        ACL_LOG_INNER_ERROR(
            "[Serialize][Dataset]Failed to TensorDatasetSerializesV2, device is %u, name is %s.", handle->devId,
            handle->name.c_str());
        itemVec.clear();
        return ret;
    }
    std::vector<std::shared_ptr<uint8_t>> ctrlSharedPtrVec;
    std::vector<rtMemQueueBuffInfo> queueBufInfoVec;
    ret = acl::TensorDataitemSerialize(itemVec, dataset->memType, queueBufInfoVec, ctrlSharedPtrVec);
    if (ret != ACL_SUCCESS) {
        ACL_LOG_INNER_ERROR(
            "[Serialize][Dataset]Failed to TensorDataitemSerialize, device is %u, name is %s.", handle->devId,
            handle->name.c_str());
        return ret;
    }

    rtMemQueueBuff_t queueBuf = {nullptr, 0U, nullptr, 0U};
    queueBuf.buffCount = queueBufInfoVec.size();
    queueBuf.buffInfo = queueBufInfoVec.data();
    ret = rtMemQueueEnQueueBuff(handle->devId, handle->qid, &queueBuf, timeout);
    if (ret == ACL_ERROR_RT_QUEUE_FULL) {
        ACL_LOG_DEBUG("queue is full, device is %u, name is %s", handle->devId, handle->name.c_str());
        return ret;
    }
    if (ret != RT_ERROR_NONE) {
        return ret;
    }
    ACL_LOG_DEBUG("success to execute acltdtSendTensor, device is %u, name is %s", handle->devId, handle->name.c_str());
    return ACL_SUCCESS;
}

aclError EnsureCurrentThreadHasContext(const acltdtChannelHandle* handle)
{
    rtContext_t rtCtx = nullptr;
    const rtError_t rtRet = rtCtxGetCurrent(&rtCtx);
    if ((rtRet != ACL_RT_SUCCESS) && (rtRet != ACL_ERROR_RT_CONTEXT_NULL)) {
        return rtRet;
    }
    if (rtCtx == nullptr) {
        if (handle->ctx_ == nullptr) {
            ACL_LOG_INFO("current thread need to create new context");
            ACL_REQUIRES_RTS_OK_WARN_NOT_SUPPORT(
                rtCtxCreateEx(&rtCtx, static_cast<uint32_t>(RT_CTX_NORMAL_MODE), static_cast<int32_t>(handle->devId)),
                rtCtxCreateEx);
            const_cast<acltdtChannelHandle*>(handle)->ctx_.reset(rtCtx, [](void* p) {
                if (p != nullptr) {
                    (void)rtCtxDestroyEx(p);
                }
            });
        }
        ACL_REQUIRES_RTS_OK_WARN_NOT_SUPPORT(rtCtxSetCurrent(handle->ctx_.get()), rtCtxSetCurrent);
    }
    return ACL_SUCCESS;
}

static size_t GetMallocSize(const size_t bufLen)
{
    // 超出当前档位就是bufLen, 在档位内就是上限值，并保存当前申请的值
    for (const size_t& size : GEAR_SIZE) {
        if (bufLen <= size) {
            return size;
        }
    }
    return bufLen;
}

aclError GetOrMallocHostMem(const acltdtChannelHandle* handle, acltdtDataset* dataset, size_t bufLen, void*& hostPtr)
{
    ACL_LOG_INFO("current need size is %zu, current mem size is %zu", bufLen, dataset->sharedMemSize_);
    ACL_REQUIRES_OK(EnsureCurrentThreadHasContext(handle));
    if (bufLen > dataset->sharedMemSize_) {
        const size_t mallocSize = GetMallocSize(bufLen);
        ACL_LOG_INFO("need mallochost size %zu, bufLen is %zu", mallocSize, bufLen);
        void* outHostAddr = nullptr;
        ACL_REQUIRES_RTS_OK_WARN_NOT_SUPPORT(
            rtMallocHost(&outHostAddr, mallocSize, acl::ACL_MODE_ID_U16), rtMallocHost);
        ACL_CHECK_MALLOC_RESULT(outHostAddr);
        dataset->sharedMem_.reset(outHostAddr, [](void* p) {
            if (p != nullptr) {
                (void)rtFreeHost(p);
            }
        });
        dataset->sharedMemSize_ = mallocSize;
    }
    hostPtr = dataset->sharedMem_.get();
    return ACL_SUCCESS;
}

aclError acltdtReceiveTensorV2(const acltdtChannelHandle* handle, acltdtDataset* dataset, int32_t timeout)
{
    ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(handle);
    size_t bufLen = 0;
    auto ret = rtMemQueuePeek(handle->devId, handle->qid, &bufLen, timeout);
    if (ret == ACL_ERROR_RT_QUEUE_EMPTY) {
        ACL_LOG_INFO("queue is empty, device is %u, name is %s", handle->devId, handle->name.c_str());
        return ret;
    }
    if (ret != RT_ERROR_NONE) {
        ACL_LOG_ERROR("peek queue [%u] failed", handle->qid);
        return ret;
    }
    ACL_LOG_INFO("peek queue [%u] success, bufLen is %zu", handle->qid, bufLen);
    if (bufLen == 0) {
        ACL_LOG_INNER_ERROR("[Check][bufLen]peek queue len cannot be zero");
        return ACL_ERROR_FAILURE;
    }
    void* hostPtr = nullptr;
    ACL_REQUIRES_OK(GetOrMallocHostMem(handle, dataset, bufLen, hostPtr));

    rtMemQueueBuff_t queueBuf = {nullptr, 0U, nullptr, 0U};
    rtMemQueueBuffInfo queueBufInfo = {hostPtr, bufLen};
    queueBuf.buffCount = 1;
    queueBuf.buffInfo = &queueBufInfo;
    ret = rtMemQueueDeQueueBuff(handle->devId, handle->qid, &queueBuf, 0);
    if (ret == ACL_ERROR_RT_QUEUE_EMPTY) {
        ACL_LOG_INFO("queue is empty, device is %u, name is %s", handle->devId, handle->name.c_str());
        return ret;
    }
    if (ret != RT_ERROR_NONE) {
        ACL_LOG_ERROR("Failed to rtMemQueueDeQueueBuf, device is %u, name is %s.", handle->devId, handle->name.c_str());
        return ret;
    }

    std::vector<acl::aclTdtDataItemInfo> itemVec;
    ret = acl::UnpackageRecvDataInfo(static_cast<uint8_t*>(hostPtr), bufLen, itemVec);
    if (ret != ACL_SUCCESS) {
        ACL_LOG_ERROR(
            "Failed to unpackage received data, device is %u, name is %s.", handle->devId, handle->name.c_str());
        return ret;
    }
    ret = acl::TensorDatasetDeserializesV2(itemVec, dataset);
    if (ret != ACL_SUCCESS) {
        ACL_LOG_INNER_ERROR(
            "[Deserialize][Dataset]Failed to deserialize tensor dataset, device is %u, name is %s.", handle->devId,
            handle->name.c_str());
        return ret;
    }
    ACL_LOG_INFO(
        "success to execute acltdtReceiveTensorV2, device is %u, name is %s.", handle->devId, handle->name.c_str());
    return ACL_SUCCESS;
}
} // namespace acl

acltdtTensorType acltdtGetTensorTypeFromItem(const acltdtDataItem* dataItem)
{
    ACL_REQUIRES_NOT_NULL_RET_INPUT_REPORT(dataItem, ACL_TENSOR_DATA_UNDEFINED);
    return dataItem->tdtType;
}

aclDataType acltdtGetDataTypeFromItem(const acltdtDataItem* dataItem)
{
    ACL_REQUIRES_NOT_NULL_RET_INPUT_REPORT(dataItem, ACL_DT_UNDEFINED);
    return dataItem->dataType;
}

void* acltdtGetDataAddrFromItem(const acltdtDataItem* dataItem)
{
    ACL_REQUIRES_NOT_NULL_RET_NULL_INPUT_REPORT(dataItem);
    if (dataItem->priorityData_ != nullptr) {
        return dataItem->priorityData_;
    }
    return dataItem->dataPtr.get();
}

size_t acltdtGetDataSizeFromItem(const acltdtDataItem* dataItem)
{
    ACL_REQUIRES_NOT_NULL_RET_INPUT_REPORT(dataItem, 0);
    return dataItem->dataLen;
}

size_t acltdtGetDimNumFromItem(const acltdtDataItem* dataItem)
{
    ACL_REQUIRES_NOT_NULL_RET_INPUT_REPORT(dataItem, 0);
    return dataItem->dims.size();
}

aclError acltdtGetSliceInfoFromItem(const acltdtDataItem* dataItem, size_t* sliceNum, size_t* sliceId)
{
    ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(dataItem);
    ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(sliceNum);
    ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(sliceId);
    *sliceNum = dataItem->sliceNum;
    *sliceId = dataItem->sliceId;
    return ACL_SUCCESS;
}

aclError acltdtGetDimsFromItem(const acltdtDataItem* dataItem, int64_t* dims, size_t dimNum)
{
    ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(dataItem);
    // check dims and dimNum
    if (dims == nullptr && dimNum != 0) {
        ACL_LOG_ERROR("[Check][Params]acltdtGetDimsFromItem failed, invalid dims and dimNum[%zu]", dimNum);
        std::string value = "nullptr/" + std::to_string(dimNum);
        acl::AclErrorLogManager::ReportInputError(
            acl::INVALID_PARAM_REASON_MSG, std::vector<const char*>({"func", "value", "param", "reason"}),
            std::vector<const char*>(
                {__func__, value.c_str(), "dims/dimNum", "If dims is nullptr, dimNum should be 0"}));
        return ACL_ERROR_INVALID_PARAM;
    }

    if (dims != nullptr && dimNum == 0) {
        ACL_LOG_ERROR("[Check][Params]acltdtGetDimsFromItem failed, invalid dims and dimNum[%zu]", dimNum);
        std::string value = std::to_string(*dims) + "/" + std::to_string(dimNum);
        acl::AclErrorLogManager::ReportInputError(
            acl::INVALID_PARAM_REASON_MSG, std::vector<const char*>({"func", "value", "param", "reason"}),
            std::vector<const char*>(
                {__func__, value.c_str(), "dims/dimNum", "If dims is not nullptr, dimNum should be greater than 0"}));
        return ACL_ERROR_INVALID_PARAM;
    }

    if (dimNum < dataItem->dims.size()) {
        ACL_LOG_ERROR(
            "[Check][dimNum]output dimNum[%zu] cannot be less than dims number[%zu]", dimNum, dataItem->dims.size());
        const std::string dimNumVal = std::to_string(dimNum);
        std::string errMsg = acl::AclErrorLogManager::FormatStr(
            "dimNum %zu cannot be less than the size of dataItem's dims %zu", dimNum, dataItem->dims.size());
        acl::AclErrorLogManager::ReportInputError(
            acl::INVALID_PARAM_REASON_MSG, std::vector<const char*>({"func", "value", "param", "reason"}),
            std::vector<const char*>({__func__, dimNumVal.c_str(), "dimNum", errMsg.c_str()}));
        return ACL_ERROR_INVALID_PARAM;
    }

    for (size_t i = 0; i < dataItem->dims.size(); ++i) {
        dims[i] = dataItem->dims[i];
    }

    return ACL_SUCCESS;
}

const char* acltdtGetDatasetName(const acltdtDataset* dataset)
{
    ACL_REQUIRES_NOT_NULL_RET_INPUT_REPORT(dataset, nullptr);
    return dataset->name.c_str();
}

acltdtDataItem* acltdtCreateDataItem(
    acltdtTensorType tdtType, const int64_t* dims, size_t dimNum, aclDataType dataType, void* data, size_t size)
{
    if (dims == nullptr && dimNum != 0) {
        ACL_LOG_ERROR("[Check][Params]acltdtCreateDataItem failed, invalid dims and dimNum[%zu]", dimNum);
        std::string value = "nullptr/" + std::to_string(dimNum);
        acl::AclErrorLogManager::ReportInputError(
            acl::INVALID_PARAM_REASON_MSG, std::vector<const char*>({"func", "value", "param", "reason"}),
            std::vector<const char*>(
                {__func__, value.c_str(), "dims/dimNum", "If dims is nullptr, dimNum should be 0"}));
        return nullptr;
    }

    if (dims != nullptr && dimNum == 0) {
        ACL_LOG_ERROR("[Check][Params]acltdtCreateDataItem failed, invalid dims and dimNum[%zu]", dimNum);
        std::string value = std::to_string(*dims) + "/" + std::to_string(dimNum);
        acl::AclErrorLogManager::ReportInputError(
            acl::INVALID_PARAM_REASON_MSG, std::vector<const char*>({"func", "value", "param", "reason"}),
            std::vector<const char*>(
                {__func__, value.c_str(), "dims/dimNum", "If dims is not nullptr, dimNum should be greater than 0"}));
        return nullptr;
    }

    constexpr size_t MAX_DIM_CNT = 128UL;
    if (dimNum > MAX_DIM_CNT) {
        ACL_LOG_ERROR(
            "[Check][Dimnum]acltdtCreateDataItem failed, dimNum[%zu] can't be larger than "
            "MAX_DIM_CNT[%zu]",
            dimNum, MAX_DIM_CNT);
        std::string expect = "less than or equal to " + std::to_string(MAX_DIM_CNT);
        const std::string dimNumVal = std::to_string(dimNum);
        acl::AclErrorLogManager::ReportInputError(
            acl::INVALID_VALUE_MSG, std::vector<const char*>({"func", "value", "param", "expect"}),
            std::vector<const char*>({__func__, dimNumVal.c_str(), "dimNum", expect.c_str()}));
        return nullptr;
    }

    if (tdtType != ACL_TENSOR_DATA_TENSOR) {
        if (dims != nullptr) {
            ACL_LOG_ERROR(
                "[Check][Dims]acltdtCreateDataItem failed, "
                "dims must be nullptr. tdtType is %d",
                tdtType);
            const std::string dimsVal = std::to_string(reinterpret_cast<uintptr_t>(dims));
            acl::AclErrorLogManager::ReportInputError(
                acl::INVALID_VALUE_MSG, std::vector<const char*>({"func", "value", "param", "expect"}),
                std::vector<const char*>({__func__, dimsVal.c_str(), "dims", "nullptr"}));
            return nullptr;
        }
        return new (std::nothrow) acltdtDataItem(tdtType, dims, dimNum, "[]", ACL_DT_UNDEFINED, "", nullptr, 0);
    }

    // tdtType: ACL_TENSOR_DATA_TENSOR
    std::string dimsStr = "[";
    acl::GetTensorDimsString(dims, dimNum, dimsStr);

    std::string typeStr;
    for (const auto& item : aclDataTypeStrMap) {
        if (item.second == dataType) {
            typeStr = item.first;
            break;
        }
    }
    std::shared_ptr<void> dataPtr;
    dataPtr.reset(data, [](const void*) {});
    return new (std::nothrow) acltdtDataItem(tdtType, dims, dimNum, dimsStr, dataType, typeStr, dataPtr, size);
}

aclError acltdtDestroyDataItem(acltdtDataItem* dataItem)
{
    ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(dataItem);
    ACL_DELETE_AND_SET_NULL(dataItem);
    return ACL_SUCCESS;
}

acltdtDataset* acltdtCreateDataset() { return new (std::nothrow) acltdtDataset(); }

aclError acltdtDestroyDataset(acltdtDataset* dataset)
{
    ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(dataset);
    ACL_DELETE_AND_SET_NULL(dataset);
    return ACL_SUCCESS;
}

aclError acltdtAddDataItem(acltdtDataset* dataset, acltdtDataItem* dataItem)
{
    ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(dataset);
    ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(dataItem);
    if (dataset->freeSelf) {
        acl::AclErrorLogManager::ReportInputError(
            acl::UNSUPPORTED_FEATURE_MSG, std::vector<const char*>({"feature", "reason"}),
            std::vector<const char*>({__func__, "item cannot be added because internal item already exists"}));
        return ACL_ERROR_FEATURE_UNSUPPORTED;
    }
    datasetMemType currentMemType = MEM_UNKNOWN;
    if (dataItem->dataPtr != nullptr) {
        rtPtrAttributes_t attr = {};
        ACL_REQUIRES_RTS_OK_WARN_NOT_SUPPORT(
            rtsPointerGetAttributes(dataItem->dataPtr.get(), &attr), rtsPointerGetAttributes);
        if ((attr.location.type == RT_MEMORY_LOC_HOST) || (attr.location.type == RT_MEMORY_LOC_UNREGISTERED)) {
            currentMemType = MEM_HOST;
        } else {
            currentMemType = MEM_DEVICE;
        }
    }
    if (dataset->memType == MEM_UNKNOWN) {
        // only MEM_UNKNOWN status can be refreshed
        dataset->memType = currentMemType;
    }

    if ((dataset->memType != MEM_UNKNOWN) && (currentMemType != MEM_UNKNOWN)) {
        if (dataset->memType != currentMemType) {
            ACL_LOG_ERROR("The memTypes in the dataset must be all host-side or device-side address");
            const std::string memTypeVal = acl::DatasetMemTypeToString(dataset->memType);
            acl::AclErrorLogManager::ReportInputError(
                acl::INVALID_PARAM_REASON_MSG, std::vector<const char*>({"func", "value", "param", "reason"}),
                std::vector<const char*>(
                    {__func__, memTypeVal.c_str(), "dataset->memType",
                     "The memTypes in the dataset must be all host-side or device-side address"}));
            return ACL_ERROR_INVALID_PARAM;
        }
    }
    dataset->blobs.push_back(dataItem);
    return ACL_SUCCESS;
}

acltdtDataItem* acltdtGetDataItem(const acltdtDataset* dataset, size_t index)
{
    ACL_REQUIRES_NOT_NULL_RET_NULL_INPUT_REPORT(dataset);
    if (index >= dataset->blobs.size()) {
        std::string errMsg = acl::AclErrorLogManager::FormatStr(
            "index %zu is greater than or equal to dataset size %zu", index, dataset->blobs.size());
        const std::string indexVal = std::to_string(index);
        acl::AclErrorLogManager::ReportInputError(
            acl::INVALID_PARAM_REASON_MSG, std::vector<const char*>({"func", "value", "param", "reason"}),
            std::vector<const char*>({__func__, indexVal.c_str(), "index", errMsg.c_str()}));
        return nullptr;
    }

    return dataset->blobs[index];
}

size_t acltdtGetDatasetSize(const acltdtDataset* dataset)
{
    ACL_REQUIRES_NOT_NULL_RET_INPUT_REPORT(dataset, 0);
    return dataset->blobs.size();
}

acltdtChannelHandle* acltdtCreateChannel(uint32_t deviceId, const char* name)
{
    ACL_REQUIRES_NOT_NULL_RET_NULL_INPUT_REPORT(name);
    static TdtHostInitFunc tdtHostInit = (TdtHostInitFunc)GetFunction("TdtHostInit");
    if (tdtHostInit == nullptr) {
        return nullptr;
    }
    const auto ret = tdtHostInit(deviceId);
    if (ret != 0) {
        ACL_LOG_INNER_ERROR("[Init][Tdt]tdt host init failed, tdt result = %d", ret);
        return nullptr;
    }
    acltdtChannelHandle* handle = new (std::nothrow) acltdtChannelHandle(deviceId, name);
    if (handle != nullptr) {
        if (!handle->recvName.empty()) {
            static TdtHostPreparePopDataFunc tdtHostPreparePopData =
                (TdtHostPreparePopDataFunc)GetFunction("TdtHostPreparePopData");
            if (tdtHostPreparePopData == nullptr) {
                return nullptr;
            }
            (void)tdtHostPreparePopData();
        }
        {
            std::unique_lock<std::mutex> lk(aclChannleMutex);
            aclChannleMap[name] = handle;
        }
    }
    return handle;
}

acltdtChannelHandle* acltdtCreateChannelWithCapacity(uint32_t deviceId, const char* name, size_t capacity)
{
    ACL_REQUIRES_NOT_NULL_RET_NULL_INPUT_REPORT(name);
    ACL_LOG_INFO("acltdtCreateChannelWithCapacity devId is %u, name is %s, capacity is %zu", deviceId, name, capacity);
    if (strnlen(name, RT_MQ_MAX_NAME_LEN) + 1 > RT_MQ_MAX_NAME_LEN) {
        ACL_LOG_ERROR("name [%s] length %zu cannot be larger than %d", name, (strlen(name) + 1U), RT_MQ_MAX_NAME_LEN);
        std::string errMsg = acl::AclErrorLogManager::FormatStr(
            "name [%s] length %zu cannot be larger than %d", name, (strlen(name) + 1U), RT_MQ_MAX_NAME_LEN);
        acl::AclErrorLogManager::ReportInputError(
            acl::INVALID_PARAM_REASON_MSG, std::vector<const char*>({"func", "value", "param", "reason"}),
            std::vector<const char*>({__func__, name, "name", errMsg.c_str()}));
        return nullptr;
    }
    acltdtChannelHandle* handle = new (std::nothrow) acltdtChannelHandle(deviceId, name);
    ACL_CHECK_MALLOC_RESULT_REPORT_RET(handle, sizeof(acltdtChannelHandle), "new", nullptr);
    handle->isTdtProcess = false;
    acltdtQueueAttr attr{};
    const size_t count = strlen(name) + 1U;
    const auto ret = memcpy_s(attr.name, RT_MQ_MAX_NAME_LEN, name, count);
    if (ret != EN_OK) {
        const std::string retCode = std::to_string(ret);
        std::stringstream ss;
        ss << std::hex << "name=0x" << reinterpret_cast<uintptr_t>(name) << ", dest=0x"
           << reinterpret_cast<uintptr_t>(attr.name) << std::dec << ", dest_max=" << RT_MQ_MAX_NAME_LEN
           << ", count=" << count << ".";
        const std::string extendInfo = ss.str();
        acl::AclErrorLogManager::ReportInputError(
            acl::STANDARD_FUNC_FAILED_MSG,
            std::vector<const char*>({"func1", "func2", "ret_code", "reason", "extend_info"}),
            std::vector<const char*>({__func__, "memcpy_s", retCode.c_str(), strerror(ret), extendInfo.c_str()}));
        ACL_LOG_ERROR(
            "[Call][MemCpy]call memcpy failed, result=%d, srcLen=%zu, dstLen=%d", ret, count, RT_MQ_MAX_NAME_LEN);
        ACL_DELETE_AND_SET_NULL(handle);
        return nullptr;
    }
    attr.depth = static_cast<uint32_t>(capacity);
    attr.workMode = RT_MQ_MODE_DEFAULT;
    attr.flowCtrlFlag = false;
    attr.flowCtrlDropTime = 0;
    attr.overWriteFlag = false;
    // queue init should be invoked when device is open
    const auto rtError = rtMemQueueInit(deviceId);
    if (rtError == ACL_ERROR_RT_FEATURE_NOT_SUPPORT) {
        ACL_LOG_INFO("queue init failed due to runtime does not support.");
        ACL_DELETE_AND_SET_NULL(handle);
        return nullptr;
    }
    if ((rtError != RT_ERROR_NONE) && (rtError != ACL_ERROR_RT_REPEATED_INIT)) {
        ACL_LOG_INNER_ERROR("queue init failed, rtError is %d", rtError);
        ACL_DELETE_AND_SET_NULL(handle);
        return nullptr;
    }
    const rtError_t rtErr = rtMemQueueCreate(static_cast<int32_t>(deviceId), &attr, &handle->qid);
    if (rtErr != RT_ERROR_NONE) {
        ACL_DELETE_AND_SET_NULL(handle);
        return nullptr;
    }
    ACL_LOG_INFO(
        "acltdtCreateChannelWithCapacity devId is %u, name is %s, real name is %s, qid is %u", deviceId,
        handle->name.c_str(), name, handle->qid);
    return handle;
}

aclError acltdtStopChannel(acltdtChannelHandle* handle)
{
    ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(handle);
    ACL_LOG_INFO("start to acltdtStopChannel, device is %u, name is %s", handle->devId, handle->name.c_str());
    if (!handle->isTdtProcess) {
        ACL_LOG_INFO("new process , stop channel is no use");
        return ACL_SUCCESS;
    }
    if (!handle->recvName.empty()) {
        static TdtHostStopFunc tdtHostStop = (TdtHostStopFunc)GetFunction("TdtHostStop");
        if (tdtHostStop == nullptr) {
            return ACL_ERROR_FAILURE;
        }
        const auto ret = tdtHostStop(handle->recvName);
        if (ret != 0) {
            ACL_LOG_INNER_ERROR(
                "[Init][Tdt]tdt host stop failed for channel %s, tdt result = %d", handle->name.c_str(), ret);
            return ACL_ERROR_FAILURE;
        }
    }
    ACL_LOG_INFO("acltdtStopChannel success, device is %u, name is %s", handle->devId, handle->name.c_str());
    return ACL_SUCCESS;
}

aclError acltdtDestroyChannel(acltdtChannelHandle* handle)
{
    ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(handle);
    ACL_LOG_INFO("start to acltdtDestroyChannel, device is %u, name is %s", handle->devId, handle->name.c_str());
    if (!handle->isTdtProcess) {
        ACL_REQUIRES_RTS_OK_WARN_NOT_SUPPORT(
            rtMemQueueDestroy(static_cast<int32_t>(handle->devId), handle->qid), rtMemQueueDestroy);
        ACL_LOG_INFO("acltdtDestroyChannel success, device is %u, name is %s", handle->devId, handle->name.c_str());
        ACL_DELETE_AND_SET_NULL(handle);
        return ACL_SUCCESS;
    }
    std::unique_lock<std::mutex> lk(aclChannleMutex);
    (void)aclChannleMap.erase(handle->name);
    if (aclChannleMap.size() == 0) {
        static TdtHostDestroyFunc tdtHostDestroy = (TdtHostDestroyFunc)GetFunction("TdtHostDestroy");
        if (tdtHostDestroy == nullptr) {
            return ACL_ERROR_FAILURE;
        }
        const auto ret = tdtHostDestroy();
        if (ret != 0) {
            ACL_LOG_INNER_ERROR("[Destroy][Tdt]TdtHostDestroy failed, tdt result = %d", ret);
        }
    }

    ACL_DELETE_AND_SET_NULL(handle);
    return ACL_SUCCESS;
}

aclError acltdtCleanChannel(acltdtChannelHandle* handle)
{
    ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(handle);
    ACL_LOG_INFO("start to acltdtCleanChannel, device is %u, name is %s", handle->devId, handle->name.c_str());
    if (!handle->isTdtProcess) {
        ACL_REQUIRES_RTS_OK_WARN_NOT_SUPPORT(rtMemQueueReset(handle->devId, handle->qid), rtMemQueueReset);
        ACL_LOG_INFO("acltdtCleanChannel success, device is %u, name is %s", handle->devId, handle->name.c_str());
        return ACL_SUCCESS;
    }
    return ACL_ERROR_FEATURE_UNSUPPORTED;
}

aclError acltdtSendTensor(const acltdtChannelHandle* handle, const acltdtDataset* dataset, int32_t timeout)
{
    ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(handle);
    ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(dataset);
    ACL_LOG_DEBUG("start to execute acltdtSendTensor, device is %u, name is %s", handle->devId, handle->name.c_str());
    if (!handle->isTdtProcess) {
        ACL_LOG_DEBUG("new process, use queue process");
        return acl::acltdtSendTensorV2(handle, dataset, timeout);
    }
    // -1 represents infinite wait, it is must be -1 now
    ACL_CHECK_INVALID_PARAM_WITH_REASON(
        timeout != -1, timeout, "Only never timeout is supported, timeout can only be set to -1");

    std::vector<tdt::DataItem> itemVec;
    const auto ret = acl::TensorDatasetSerializes(dataset, itemVec);
    if (ret != ACL_SUCCESS) {
        ACL_LOG_INNER_ERROR(
            "[Serialize][Dataset]Failed to TensorDatasetSerializes, device is %u, name is %s.", handle->devId,
            handle->name.c_str());
        itemVec.clear();
        return ret;
    }

    static TdtHostPushDataFunc tdtHostPushData = (TdtHostPushDataFunc)GetFunction("TdtHostPushData");
    if (tdtHostPushData == nullptr) {
        return ACL_ERROR_FAILURE;
    }
    int32_t sendRet = tdtHostPushData(handle->name, itemVec, 0);
    if (sendRet != 0) {
        ACL_LOG_INNER_ERROR(
            "[Push][Data]Failed to push data, tdt result = %d, device is %u, name is %s.", sendRet, handle->devId,
            handle->name.c_str());
        return ACL_ERROR_FAILURE;
    }

    ACL_LOG_DEBUG("success to execute acltdtSendTensor, device is %u, name is %s", handle->devId, handle->name.c_str());
    return ACL_SUCCESS;
}

aclError acltdtReceiveTensor(const acltdtChannelHandle* handle, acltdtDataset* dataset, int32_t timeout)
{
    ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(handle);
    ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(dataset);
    ACL_LOG_INFO("start to execute acltdtReceiveTensor, device is %u, name is %s", handle->devId, handle->name.c_str());
    if (!handle->isTdtProcess) {
        ACL_LOG_INFO("new process, use queue process");
        return acl::acltdtReceiveTensorV2(handle, dataset, timeout);
    }
    // -1 represents infinite wait, it is must be -1 now
    ACL_CHECK_INVALID_PARAM_WITH_REASON(
        timeout != -1, timeout, "Only never timeout is supported, timeout can only be set to -1");

    if (handle->recvName.empty()) {
        ACL_LOG_ERROR(
            "[Check][Recvname]it is not a receive channel, failed to receive, device is %u, name is %s", handle->devId,
            handle->name.c_str());
        return ACL_ERROR_INVALID_PARAM;
    }

    std::vector<tdt::DataItem> itemVec;
    static TdtHostPopDataFunc tdtHostPopData = (TdtHostPopDataFunc)GetFunction("TdtHostPopData");
    if (tdtHostPopData == nullptr) {
        return ACL_ERROR_FAILURE;
    }
    const int32_t recvRet = tdtHostPopData(handle->recvName, itemVec);
    if (recvRet != 0) {
        ACL_LOG_INNER_ERROR(
            "[Pop][Data]Failed to receive, tdt result = %d, device is %u, name is %s.", recvRet, handle->devId,
            handle->name.c_str());
        return ACL_ERROR_FAILURE;
    }

    const auto ret = acl::TensorDatasetDeserializes(itemVec, dataset);
    if (ret != ACL_SUCCESS) {
        ACL_LOG_INNER_ERROR(
            "[Deserialize][Dataset]Failed to TensorDatasetDeserializes, device is %u, name is %s.", handle->devId,
            handle->name.c_str());
        return ret;
    }

    ACL_LOG_INFO(
        "success to execute acltdtReceiveTensor, device is %u, name is %s", handle->devId, handle->name.c_str());
    return ACL_SUCCESS;
}

aclError acltdtQueryChannelSize(const acltdtChannelHandle* handle, size_t* size)
{
    ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(handle);
    ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(size);
    if (handle->isTdtProcess) {
        ACL_LOG_DEBUG("acltdtQueryChannelSize is not supported");
        return ACL_ERROR_FEATURE_UNSUPPORTED;
    }
    ACL_LOG_DEBUG("start to execute acltdtQueryChannelSize, device is %u, qid is %u", handle->devId, handle->qid);
    rtMemQueueInfo_t info;
    ACL_REQUIRES_RTS_OK(rtMemQueueQueryInfo(static_cast<int32_t>(handle->devId), handle->qid, &info));
    *size = static_cast<size_t>(info.size);
    ACL_LOG_DEBUG(
        "success to execute acltdtQueryChannelSize, size is %zu, device is %u, qid is %u", *size, handle->devId,
        handle->qid);
    return ACL_SUCCESS;
}