/*
 * Copyright (c) Huawei Technologies Co., Ltd. 2025. All rights reserved.

 * ubs-io is licensed under the Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *      http://license.coscl.org.cn/MulanPSL2
 * 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 FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 */

#include "bio_client_agent.h"
#include <dlfcn.h>
#include <iostream>
#include "bio_client_log.h"
#include "bio_client_net.h"
#include "bio_crc_util.h"
#include "bio_functions.h"
#include "bio_trace.h"
#include "bio_tracepoint_helper.h"
#include "message_op.h"

using namespace ock::bio;
using namespace ock::bio::agent;

BResult BioClientAgent::Initialize(WorkerMode mode)
{
    mMode = mode;
    if (mMode == CONVERGENCE) {
#ifdef DEBUG_UT
        const char *soFileName = "libbio_server.so";
        handler = dlopen(soFileName, RTLD_NOW);
#else
        std::string soFileName = "/usr/lib64/libbio_server.so";
        char *canonicalPath = realpath(soFileName.c_str(), nullptr);
        if (canonicalPath == nullptr) {
            CLIENT_LOG_ERROR("Failed to open library, not exist, " << soFileName << ".");
            return BIO_NOT_EXISTS;
        }

        handler = dlopen(canonicalPath, RTLD_NOW);
        free(canonicalPath);
        canonicalPath = nullptr;
#endif
        if (handler == nullptr) {
            CLIENT_LOG_ERROR("Failed to open library() " << soFileName << " dlopen , error " << dlerror());
            return BIO_INNER_ERR;
        }

        if (InitOperation() != BIO_OK) {
            CLIENT_LOG_ERROR("Failed to init operation.");
            dlclose(handler);
            return BIO_INNER_ERR;
        }

        // Start boostio server for converged deployment mode
        int32_t ret = BIO_INNER_ERR;
        BIO_TP_START(SDK_BIO_AGENT_START_OP_FAIL, &ret, BIO_INNER_ERR);
        ret = startOp();
        BIO_TP_END;
        if (ret != BIO_OK) {
            CLIENT_LOG_ERROR("Failed to start bio server, ret:" << ret << ".");
            dlclose(handler);
            return BIO_INNER_ERR;
        }
    }
    return BIO_OK;
}

void BioClientAgent::Exit()
{
    if (mMode == CONVERGENCE) {
        exitOp();
    }
}

BResult BioClientAgent::InitUpgradeOperation()
{
    if ((notifyUpdateOp = reinterpret_cast<NotifyUpdateFuncPtr>(LoadFunction("NotifyUpdate"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    if ((checkUpdateReadyOp = reinterpret_cast<CheckUpdateReadyFuncPtr>(LoadFunction("CheckUpdateReady"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    return BIO_OK;
}

BResult BioClientAgent::InitOperation()
{
    if ((startOp = reinterpret_cast<BioServerStartFuncPtr>(LoadFunction("BioServerInit"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    if ((exitOp = reinterpret_cast<BioServerExitFuncPtr>(LoadFunction("BioServerExit"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    if ((getCrcFlag = reinterpret_cast<GetBioServerCrcFlagFuncPtr>(LoadFunction("GetCrcFlag"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    if ((getCliFlag = reinterpret_cast<GetBioServerCliFlagFuncPtr>(LoadFunction("GetCliFlag"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    if ((getPrometheusToggle =
             reinterpret_cast<GetBioServerPromethuesToggleFuncPtr>(LoadFunction("GetPrometheusToggle"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    if ((getListenAddress = reinterpret_cast<GetBioServerListenAddressFuncPtr>(
             LoadFunction("GetPrometheusListenAddress"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    if ((getTimeOut = reinterpret_cast<GetBioServertimeOutFuncPtr>(LoadFunction("GetNegoWorkIoTimeOut"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    if ((getScrapeIntervalSec = reinterpret_cast<GetBioServerScrapeIntervalSecFuncPtr>(
             LoadFunction("GetPrometheusScrapeIntervalSec"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    if ((getNetEngineOp = reinterpret_cast<GetBioServerNetEngineFuncPtr>(LoadFunction("GetBioServerNet"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    if ((getLocalNidOp = reinterpret_cast<GetLocalNidFuncPtr>(LoadFunction("GetLocalNid"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    if ((getQuotaInfoOp = reinterpret_cast<GetQuotaInfoFuncPtr>(LoadFunction("GetQuotaInfo"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    if ((allocQuotaOp = reinterpret_cast<AllocQuotaFuncPtr>(LoadFunction("AllocQuota"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    if ((freeQuotaOp = reinterpret_cast<FreeQuotaFuncPtr>(LoadFunction("FreeQuota"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    if ((getNodeViewOp = reinterpret_cast<GetNodeViewFuncPtr>(LoadFunction("GetNodeView"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    if ((getPtViewOp = reinterpret_cast<GetPtViewFuncPtr>(LoadFunction("GetPtView"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    if ((createFlowMasterOp = reinterpret_cast<CreateFlowMasterFuncPtr>(LoadFunction("CreateFlowMaster"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    if ((createFlowSlaveOp = reinterpret_cast<CreateFlowSlaveFuncPtr>(LoadFunction("CreateFlowSlave"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    if ((destroyFlowOp = reinterpret_cast<DestroyFlowFuncPtr>(LoadFunction("DestroyFlow"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    if ((getSliceOp = reinterpret_cast<GetSliceFuncPtr>(LoadFunction("GetSlice"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    if ((putOp = reinterpret_cast<PutFuncPtr>(LoadFunction("Put"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    if ((getOp = reinterpret_cast<GetFuncPtr>(LoadFunction("Get"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    if ((deleteOp = reinterpret_cast<DeleteFuncPtr>(LoadFunction("Delete"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    if ((addDiskOp = reinterpret_cast<AddDiskFuncPtr>(LoadFunction("AddDisk"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    if ((listOp = reinterpret_cast<ListFuncPtr>(LoadFunction("List"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    if ((statOp = reinterpret_cast<StatFuncPtr>(LoadFunction("Stat"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    if ((loadOp = reinterpret_cast<LoadFuncPtr>(LoadFunction("Load"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    if ((cacheHitOp = reinterpret_cast<GetCacheHitLocalFuncPtr>(LoadFunction("GetCacheHitLocal"))) == nullptr) {
        return BIO_INNER_ERR;
    }
    if ((cacheResourceOp = reinterpret_cast<CalcCacheResourceLocalFuncPtr>(LoadFunction("CalcCacheResourceLocal"))) ==
        nullptr) {
        return BIO_INNER_ERR;
    }
    if ((getTracePointsOp = reinterpret_cast<GetTracePointsLocalFuncPtr>(LoadFunction("GetTracePointsLocal"))) ==
        nullptr) {
        return BIO_INNER_ERR;
    }

    return InitUpgradeOperation();
}

void *BioClientAgent::LoadFunction(const char *name)
{
    void *ptr = nullptr;
    BIO_TP_START(SDK_BIO_AGENT_LOAD_FUNC_FAIL, &ptr, nullptr);
    ptr = dlsym(handler, name);
    BIO_TP_END;
    if (ptr == nullptr) {
        CLIENT_LOG_ERROR("Failed to load function " << name);
        return nullptr;
    }
    return ptr;
}

BResult BioClientAgent::SendGetLocalNodeInfoRequest(uint16_t &protocol, CmNodeId &localNid)
{
    GetLocalNidRequest req = {{MESSAGE_MAGIC, 0, 0, 0, getpid()}};
    GetLocalNidResponse rsp;
    auto ret = net::BioClientNet::Instance()->SendSync<GetLocalNidRequest, GetLocalNidResponse>(
        INVALID_NID, BIO_OP_SDK_GET_NODE_INFO, req, rsp);
    if (ret != BIO_OK) {
        return ret;
    }
    // 集群组仅有一个(groupId为0).
    if (rsp.groupId != 0) {
        return BIO_INNER_ERR;
    }
    protocol = rsp.protocol;
    localNid = {rsp.groupId, rsp.nodeId};
    return BIO_OK;
}

BResult BioClientAgent::SendGetNodeInfoRequest(uint16_t masterPtId, uint16_t slavePtId, FileLocationQueryRsp &rsp)
{
    FileLocationQueryReq req = {masterPtId, slavePtId};
    return net::BioClientNet::Instance()->SendSync<FileLocationQueryReq, FileLocationQueryRsp>(
        INVALID_NID, BIO_OP_SDK_GET_NODE_INFO_BY_PT, req, rsp);
}

bool BioClientAgent::GetConfigCrcFlag()
{
    return getCrcFlag();
}

bool BioClientAgent::GetConfigCliFlag()
{
    return getCliFlag();
}

bool BioClientAgent::GetConfigPrometheusToggle()
{
    return getPrometheusToggle();
}

const char *BioClientAgent::GetPrometheusListenAddress()
{
    return getListenAddress();
}

uint32_t BioClientAgent::GetNegoWorkIoTimeOut()
{
    return getTimeOut();
}

uint32_t BioClientAgent::GetPrometheusScrapeIntervalSec()
{
    return getScrapeIntervalSec();
}

BResult BioClientAgent::GetLocalNodeInfo(uint16_t &protocol, CmNodeId &localNid)
{
    BResult ret = BIO_OK;
    if (mMode == CONVERGENCE) {
        GetLocalNidResponse getLocalNidRsp{};
        ret = getLocalNidOp(&getLocalNidRsp);
        localNid = {getLocalNidRsp.groupId, getLocalNidRsp.nodeId};
        mLocalNid = localNid;
        protocol = getLocalNidRsp.protocol;
    } else {
        ret = SendGetLocalNodeInfoRequest(protocol, localNid);
        mLocalNid = localNid;
    }
    return ret;
}

BResult BioClientAgent::GetLocalQuotaInfo(uint32_t scene, bool &enable, uint64_t &preloadSize)
{
    BResult ret = BIO_OK;
    QueryQuotaRequest req = {{MESSAGE_MAGIC, 0, 0, 0, getpid()}};
    QueryQuotaResponse rsp;
    BIO_TP_START(NO_PROCESS_GET_LOCAL_QUOTA, 0);
    if (mMode == CONVERGENCE) {
        ret = getQuotaInfoOp(&req, &rsp);
    } else {
        ret = net::BioClientNet::Instance()->SendSync<QueryQuotaRequest, QueryQuotaResponse>(
            INVALID_NID, BIO_OP_SDK_GET_QUOTA_INFO, req, rsp);
    }
    if (ret == BIO_OK) {
        if (rsp.preloadSize > NO_128 * IO_SIZE_1M) { // quota初始预取大小最大不超过128MB.
            CLIENT_LOG_ERROR("Quota preload size too large, preloadSize " << rsp.preloadSize << ".");
            return BIO_INNER_ERR;
        }
        enable = rsp.enable;
        preloadSize = rsp.preloadSize;
    }
    BIO_TP_END;
    return ret;
}

BResult BioClientAgent::AllocQuota(AllocQuotaRequest &req, uint64_t &expectPreloadSize)
{
    BResult ret = BIO_INNER_ERR;
    AllocQuotaResponse rsp = {0};
    if (mMode == CONVERGENCE) {
        ret = allocQuotaOp(&req, &rsp);
    } else {
        ret = net::BioClientNet::Instance()->SendSync<AllocQuotaRequest, AllocQuotaResponse>(
            INVALID_NID, BIO_OP_SDK_ALLOC_QUOTA, req, rsp);
    }
    expectPreloadSize = std::min<uint64_t>(expectPreloadSize, rsp.exceptQuota);
    if (expectPreloadSize > NO_1024 * IO_SIZE_1M) {
        return BIO_ERR;
    }
    return ret;
}

BResult BioClientAgent::FreeQuota(FreeQuotaRequest &req)
{
    BResult ret = BIO_INNER_ERR;
    if (mMode == CONVERGENCE) {
        ret = freeQuotaOp(&req);
    } else {
        BResult hdlRet = BIO_INNER_ERR;
        ret = net::BioClientNet::Instance()->SendSync<FreeQuotaRequest, BResult>(INVALID_NID, BIO_OP_SDK_FREE_QUOTA,
                                                                                 req, hdlRet);
        if (ret == BIO_OK && hdlRet != BIO_OK) {
            ret = hdlRet;
        }
    }
    return ret;
}

BResult BioClientAgent::GetClusterNodeView(uint64_t &curNodeTimes,
                                           std::map<CmNodeId, CmNodeInfo, CmNodeIdCmp> &nodeView)
{
    BResult ret = BIO_OK;
    int32_t flag = 0;
    uint32_t progressBar = 0;
    static uint32_t maxRetryCnt = NO_512;
    uint32_t retryCnt = 0;

    do {
        QueryNodeViewRequest req = {{MESSAGE_MAGIC, 0, 0, 0, getpid()}, progressBar};
        QueryNodeViewResponse rsp;
        if (mMode == CONVERGENCE) {
            ret = getNodeViewOp(&req, &rsp);
        } else {
            ret = net::BioClientNet::Instance()->SendSync<QueryNodeViewRequest, QueryNodeViewResponse>(
                INVALID_NID, BIO_OP_SDK_GET_NODE_VIEW, req, rsp);
        }
        if (ret != BIO_OK) {
            nodeView.clear();
            return ret;
        }
        if (rsp.flag == 0) { // 此处没有获取到视图表示已经完成, 直接退出循环
            break;
        }

        if (rsp.num > CLUSTER_NODE_SIZE) { // 校验集群的节点数
            nodeView.clear();
            return BIO_INNER_RETRY;
        }
        for (uint32_t i = 0; i < rsp.num; i++) {
            if (rsp.desc[i].num > DISK_MAX_SIZE) { // 校验节点上的磁盘数
                nodeView.clear();
                return BIO_INNER_RETRY;
            }
            std::vector<CmDiskInfo> disks;
            for (uint32_t j = 0; j < rsp.desc[i].num; j++) {
                disks.push_back(
                    {rsp.desc[i].diskDesc[j].diskId, static_cast<CmDiskStatus>(rsp.desc[i].diskDesc[j].diskStatus)});
            }
            rsp.desc[i].ip[IP_MAX_SIZE - 1] = '\0';
            nodeView.insert(
                std::make_pair(CmNodeId(rsp.desc[i].groupId, rsp.desc[i].nodeId),
                               CmNodeInfo(CmNodeId(rsp.desc[i].groupId, rsp.desc[i].nodeId), rsp.desc[i].ip,
                                          rsp.desc[i].port, static_cast<CmNodeStatus>(rsp.desc[i].status), disks)));
        }
        flag = rsp.flag;
        progressBar += rsp.num;
        curNodeTimes = rsp.curNodeTimes;
        if ((retryCnt++) > maxRetryCnt) { // 限制最大512次分段获取集群视图, 因为在512次内必定可以获取完整的集群视图.
            break;
        }
    } while (flag == 1);

    return BIO_OK;
}

BResult BioClientAgent::GetPtView(uint64_t &curPtTimes, std::map<uint16_t, CmPtInfo> &ptView)
{
    BResult ret = BIO_OK;
    int32_t flag = 0;
    uint32_t progressBar = 0;
    static uint32_t maxRetryCnt = NO_1024;
    uint32_t retryCnt = 0;
    do {
        QueryPtViewRequest req = {{MESSAGE_MAGIC, 0, 0, 0, getpid()}, progressBar};
        QueryPtViewResponse rsp;
        if (mMode == CONVERGENCE) {
            ret = getPtViewOp(&req, &rsp);
        } else {
            ret = net::BioClientNet::Instance()->SendSync<QueryPtViewRequest, QueryPtViewResponse>(
                INVALID_NID, BIO_OP_SDK_QUERY_PT_VIEW, req, rsp);
        }
        if (ret != BIO_OK) {
            ptView.clear();
            return ret;
        }
        if (rsp.flag == 0) { // 此处没有获取到视图表示已经完成, 直接退出循环
            break;
        }

        if (rsp.num > PT_SIZE || rsp.copyNum > PT_COPY_MAX_SIZE) {
            ptView.clear();
            return BIO_INNER_RETRY;
        }
        for (uint32_t i = 0; i < rsp.num; i++) {
            std::vector<CmPtCopy> copys;
            for (uint32_t j = 0; j < rsp.copyNum; j++) {
                copys.push_back({rsp.desc[i].copys[j].nodeId, rsp.desc[i].copys[j].diskId,
                                 static_cast<CmCopyState>(rsp.desc[i].copys[j].state)});
            }
            ptView.insert(std::make_pair(
                rsp.desc[i].ptId,
                CmPtInfo(rsp.desc[i].version, rsp.desc[i].ptId, static_cast<CmPtState>(rsp.desc[i].state),
                         rsp.desc[i].masterNodeId, rsp.desc[i].masterDiskId, copys)));
        }
        flag = rsp.flag;
        progressBar += rsp.num;
        curPtTimes = rsp.curPtTimes;
        if ((retryCnt++) > maxRetryCnt) { // 限制最大1024次分段获取视图, 因为在1024次内必定可以获取完整的分区视图.
            break;
        }
    } while (flag == 1);

    return BIO_OK;
}

BResult BioClientAgent::SendCreateFlowRequestLocal(CmPtInfo &ptEntry, uint16_t ptId, uint16_t opType, uint64_t &flowId,
                                                   bool &isDegrade)
{
    BResult ret = BIO_OK;
    CreateFlowRequest req;
    if (opType == 0) {
        req = {{MESSAGE_MAGIC, ptId, ptEntry.version, mLocalNid.VNodeId(), getpid()}, opType, 0, false};
    } else if (opType == 1) {
        req = {{MESSAGE_MAGIC, ptId, ptEntry.version, mLocalNid.VNodeId(), getpid()}, opType, flowId, isDegrade};
    }
    CreateFlowResponse rsp;
    static uint32_t createFlowTimeout = NO_60;
    uint64_t startTime = Monotonic::TimeSec();
    do {
        ret = net::BioClientNet::Instance()->SendSync<CreateFlowRequest, CreateFlowResponse>(
            INVALID_NID, BIO_OP_SDK_CREATE_FLOW, req, rsp);
        uint64_t retryTime = Monotonic::TimeSec() - startTime;
        if (UNLIKELY(ret == BIO_NOT_READY && retryTime < createFlowTimeout)) {
            CLIENT_LOG_WARN("Remote cache service not ready, need retry, ret:"
                            << ret << ", nodeId:" << mLocalNid.VNodeId() << ", ptId:" << ptId << ".");
            sleep(NO_3);
        } else {
            break;
        }
    } while (ret == BIO_NOT_READY);
    if (UNLIKELY(ret != BIO_OK)) {
        CLIENT_LOG_ERROR("Send sync create flow request failed, ret:" << ret << ", nodeId:" << mLocalNid.VNodeId()
                                                                      << ", ptId:" << ptId << ".");
        return ret;
    }

    if (opType == 0) {
        flowId = rsp.flowId;
        isDegrade = rsp.isDegrade;
    } else if (opType == 1 && rsp.flowId != 0) {
        ret = BIO_ERR;
    }
    return ret;
}

BResult BioClientAgent::SendDestroyFlowRequestLocal(CmPtInfo &ptEntry, uint16_t ptId, uint64_t flowId)
{
    DestroyFlowRequest req = {{MESSAGE_MAGIC, ptId, ptEntry.version, mLocalNid.VNodeId(), getpid()}, flowId};
    DestroyFlowResponse rsp;
    BResult ret = net::BioClientNet::Instance()->SendSync<DestroyFlowRequest, DestroyFlowResponse>(
        INVALID_NID, BIO_OP_SDK_DESTROY_FLOW, req, rsp);
    if (UNLIKELY(ret != BIO_OK)) {
        CLIENT_LOG_ERROR("Send sync destroy flow request failed:" << ret << ", nodeId:" << mLocalNid.VNodeId()
                                                                  << ", ptId:" << ptId << ", flowId:" << flowId << ".");
        return ret;
    }

    return BIO_OK;
}

BResult BioClientAgent::CreateFlowLocal(pid_t procId, CmPtInfo &ptEntry, FlowInfo &flowInfo)
{
    if (mMode == CONVERGENCE) {
        if (flowInfo.opType == 0) {
            CreateFlowRequest req = {{MESSAGE_MAGIC, flowInfo.ptId, ptEntry.version, mLocalNid.VNodeId(), procId},
                                     flowInfo.opType,
                                     0,
                                     false};
            CreateFlowResponse rsp;
            auto ret = createFlowMasterOp(&req, &rsp);
            flowInfo.flowId = rsp.flowId;
            flowInfo.isDegrade = rsp.isDegrade;
            return ret;
        } else {
            CreateFlowRequest req = {{MESSAGE_MAGIC, flowInfo.ptId, ptEntry.version, mLocalNid.VNodeId(), procId},
                                     flowInfo.opType,
                                     flowInfo.flowId,
                                     flowInfo.isDegrade};
            return createFlowSlaveOp(&req);
        }
    } else {
        return SendCreateFlowRequestLocal(ptEntry, flowInfo.ptId, flowInfo.opType, flowInfo.flowId, flowInfo.isDegrade);
    }
}

BResult BioClientAgent::DestroyFlowLocal(pid_t procId, CmPtInfo &ptEntry, uint16_t ptId, uint64_t flowId)
{
    if (mMode == CONVERGENCE) {
        DestroyFlowRequest req = {{MESSAGE_MAGIC, ptId, ptEntry.version, mLocalNid.VNodeId(), getpid()}, flowId};
        return destroyFlowOp(&req);
    } else {
        return SendDestroyFlowRequestLocal(ptEntry, ptId, flowId);
    }
}

bool BioClientAgent::CheckGetSliceRsp(GetSliceResponse **rsp)
{
    if (((*rsp)->addrNum > SLICE_ADDR_MAX_SIZE) || ((*rsp)->sliceLen >= NO_512)) {
        return false;
    }
    for (uint32_t i = 0; i < (*rsp)->addrNum; i++) {
        if ((*rsp)->addr[i].chunkLen > IO_SIZE_4M) {
            return false;
        }
    }
    return true;
}

BResult BioClientAgent::SendPrepareResourceLocal(CmPtInfo &ptEntry, uint64_t flowId, uint64_t offset, uint64_t index,
                                                 uint64_t length, GetSliceResponse **rsp)
{
    GetSliceRequest req = {
        {MESSAGE_MAGIC, ptEntry.ptId, ptEntry.version, mLocalNid.VNodeId(), getpid()}, flowId, offset, index, length};
    uint64_t rspLen = 0;
    auto ret = net::BioClientNet::Instance()->SendSync<GetSliceRequest, GetSliceResponse>(
        INVALID_NID, BIO_OP_SDK_GET_SLICE, req, rsp, rspLen);
    if (ret != BIO_OK) {
        return ret;
    } else {
        if (*rsp == nullptr) {
            return BIO_INNER_ERR;
        }
        if (rspLen < sizeof(GetSliceResponse) || rspLen - sizeof(GetSliceResponse) < (*rsp)->sliceLen) {
            delete[] static_cast<uint8_t *>(static_cast<void *>(*rsp));
            return BIO_INVALID_PARAM;
        }
    }
    if (!CheckGetSliceRsp(rsp)) {
        delete[] static_cast<uint8_t *>(static_cast<void *>(*rsp));
        return BIO_INNER_ERR;
    }
    return BIO_OK;
}

BResult BioClientAgent::PrepareResource(CmPtInfo &ptEntry, uint64_t flowId, uint64_t offset, uint64_t index,
                                        uint64_t length, GetSliceResponse **rsp)
{
    if (mMode == CONVERGENCE) {
        GetSliceRequest req = {{MESSAGE_MAGIC, ptEntry.ptId, ptEntry.version, mLocalNid.VNodeId(), getpid()},
                               flowId,
                               offset,
                               index,
                               length};
        return getSliceOp(&req, rsp);
    } else {
        return SendPrepareResourceLocal(ptEntry, flowId, offset, index, length, rsp);
    }
}

void BioClientAgent::SendPutRequestLocal(PutRequest *req, Callback &callback)
{
    net::BioClientNet::Instance()->SendAsyncBuff(INVALID_NID, BIO_OP_SDK_PUT, static_cast<void *>(req),
                                                 sizeof(PutRequest) + req->sliceLen, callback);
}

void BioClientAgent::PutLocal(PutRequest *req, Callback &callback)
{
    if (mMode == CONVERGENCE) {
        PutResponse rsp;
        BIO_TRACE_START(SDK_TRACE_PUT_LOCAL_SYNC);
        auto ret = putOp(req, &rsp);
        BIO_TRACE_END(SDK_TRACE_PUT_LOCAL_SYNC, ret);
        callback.cb(callback.cbCtx, &rsp, sizeof(PutResponse), ret);
    } else {
        BIO_TRACE_START(SDK_TRACE_PUT_LOCAL_SYNC);
        SendPutRequestLocal(req, callback);
        BIO_TRACE_END(SDK_TRACE_PUT_LOCAL_SYNC, BIO_OK);
    }
}

BResult BioClientAgent::SendGetRequestLocal(GetRequest &req, char *value, uint64_t &realLen)
{
    GetResponse rsp;
    auto ret = net::BioClientNet::Instance()->SendSync<GetRequest, GetResponse>(INVALID_NID, BIO_OP_SDK_GET, req, rsp);
    if (UNLIKELY(ret != BIO_OK)) {
        CLIENT_LOG_ERROR("Send sync get request failed, ret:" << ret << ", key:" << req.key << ", offset:" << req.offset
                                                              << ", length:" << req.length
                                                              << ", dstNid:" << mLocalNid.VNodeId() << ".");
    } else {
        if (rsp.num > SLICE_ADDR_SIZE) {
            return BIO_INVALID_PARAM;
        }
        realLen = rsp.realLen;
        if (realLen > req.length) {
            return BIO_INNER_ERR;
        }
        uint64_t off = 0;
        uint64_t cpyLength = req.length;
        for (uint32_t idx = 0; idx < rsp.num; idx++) {
            if (rsp.addrLen[idx] > BIO_IO_MAX_LEN) {
                return BIO_INVALID_PARAM;
            }
            uint8_t *addr = net::BioClientNet::Instance()->GetShmAddress(rsp.addrOffset[idx], rsp.addrLen[idx]);
            if (addr == nullptr) {
                CLIENT_LOG_ERROR("Send sync request get shm addr failed.");
                break;
            }
            ret =
                memcpy_s(static_cast<void *>(value + off), cpyLength, reinterpret_cast<void *>(addr), rsp.addrLen[idx]);
            if (UNLIKELY(ret != 0)) {
                CLIENT_LOG_ERROR("Memory copy data to user failed, ret:" << ret << ", idx:" << idx
                                                                         << ", len:" << rsp.addrLen[idx] << ".");
                break;
            }
            off += rsp.addrLen[idx];
            cpyLength -= rsp.addrLen[idx];
        }

        if (rsp.isAlloc) {
            FreeMemRequest freeReq = {req.comm, rsp.num, 0, {0}};
            for (uint32_t idx = 0; idx < rsp.num; idx++) {
                freeReq.addr[idx] = rsp.addrOffset[idx];
            }
            auto freeRet =
                net::BioClientNet::Instance()->SendAsync<FreeMemRequest>(INVALID_NID, BIO_OP_SDK_FREE_MEM, freeReq);
            if (freeRet != BIO_OK) {
                CLIENT_LOG_ERROR("Send async free request failed, ret:" << ret << ".");
            }
        }
        if (req.enableCrc && ret == BIO_OK) {
            uint32_t currentCrc = BioCrcUtil::Crc32(value, rsp.realLen);
            if (rsp.dataCrc != currentCrc) {
                CLIENT_LOG_ERROR("Client get failed to verify the CRC, << key:"
                                 << req.key << ", origin crc:" << rsp.dataCrc << ", current crc:" << currentCrc);
                ret = BIO_CRC_ERR;
            }
        }
    }
    return ret;
}

BResult BioClientAgent::GetLocal(GetRequest &req, char *value, uint64_t &realLen)
{
    req.size = req.length;
    if (mMode == CONVERGENCE) {
        req.isMr = 0;
        req.address = reinterpret_cast<uintptr_t>(value);
        GetResponse rsp;
        auto ret = getOp(&req, &rsp);
        if (UNLIKELY(ret != BIO_OK)) {
            return ret;
        }

        realLen = rsp.realLen;
        if (realLen > req.length) {
            return BIO_INNER_ERR;
        }
        realLen = rsp.realLen;
        if (req.enableCrc && ret == BIO_OK) {
            uint32_t currentCrc = BioCrcUtil::Crc32(value, rsp.realLen);
            if (rsp.dataCrc != currentCrc) {
                CLIENT_LOG_ERROR("Client get failed to verify the CRC, key:"
                                 << req.key << ", origin crc:" << rsp.dataCrc << ", current crc:" << currentCrc << ".");
                ret = BIO_CRC_ERR;
            }
        }
        return ret;
    } else {
        return SendGetRequestLocal(req, value, realLen);
    }
}

void BioClientAgent::SendDeleteRequestLocal(DeleteRequest &req, Callback &callback)
{
    net::BioClientNet::Instance()->SendAsync<DeleteRequest>(INVALID_NID, BIO_OP_SDK_DELETE, req, callback);
}

void BioClientAgent::DeleteLocal(DeleteRequest &req, Callback &callback)
{
    if (mMode == CONVERGENCE) {
        auto ret = deleteOp(&req);
        callback.cb(callback.cbCtx, &ret, sizeof(ret), BIO_OK);
    } else {
        return SendDeleteRequestLocal(req, callback);
    }
}

BResult BioClientAgent::AddDisk(AddDiskRequest &req, AddDiskResponse &rsp)
{
    BIO_TP_START(SDK_ADD_DISK_BY_SEPARATES, 0)
    if (mMode == CONVERGENCE) {
        return addDiskOp(&req, &rsp);
    }
    BIO_TP_END;
    return SendAddDiskRequest(req, rsp);
}

BResult BioClientAgent::SendAddDiskRequest(AddDiskRequest &req, AddDiskResponse &rsp)
{
    return net::BioClientNet::Instance()->SendSync<AddDiskRequest, AddDiskResponse>(INVALID_NID, BIO_OP_SDK_ADD_DISK,
                                                                                    req, rsp);
}

BResult BioClientAgent::CallServerListIntf(ListRequest &req, std::unordered_map<std::string, ObjStat> &objs)
{
    req.address = 0;
    req.size = 0;
    req.mrKey = 0;
    ListResponse *rsp = nullptr;
    auto ret = listOp(&req, &rsp);
    if (ret != BIO_OK) {
        return ret;
    }

    if (rsp->num != 0) {
        auto statBuff = static_cast<ObjStat *>(static_cast<void *>(rsp->statBuf));
        for (uint32_t i = 0; i < rsp->num; i++) {
            if (objs.size() >= 1000U) {
                break;
            }
            ObjStat stat;
            CopyKey(stat.key, statBuff[i].key, KEY_MAX_SIZE);
            stat.size = statBuff[i].size;
            stat.time = statBuff[i].time;
            objs.insert({stat.key, stat});
        }
    }
    delete[] rsp;
    return BIO_OK;
}

BResult BioClientAgent::SendListRequestLocal(ListRequest &req, std::unordered_map<std::string, ObjStat> &objs)
{
    req.address = 0;
    req.size = 0;
    req.mrKey = 0;
    ListResponse rsp;
    BResult ret =
        net::BioClientNet::Instance()->SendSync<ListRequest, ListResponse>(INVALID_NID, BIO_OP_SDK_LIST, req, rsp);
    if (ret != BIO_OK) {
        CLIENT_LOG_ERROR("Send sync list request failed, ret:" << ret << ", prefix:" << req.prefix << ".");
        return ret;
    }
    if (UNLIKELY(rsp.num > 1000U || rsp.buffLen != 0)) {
        return BIO_INNER_RETRY;
    }

    if (rsp.num != 0) {
        uint8_t *addr = net::BioClientNet::Instance()->GetShmAddress(rsp.addrOffset, (rsp.num * sizeof(ObjStat)));
        if (addr == nullptr) {
            CLIENT_LOG_ERROR("Send list request get shm addr failed.");
            return BIO_INNER_ERR;
        }
        auto statInfo = static_cast<ObjStat *>(static_cast<void *>(addr));
        for (uint32_t i = 0; i < rsp.num; i++) {
            if (objs.size() >= 1000U) {
                break;
            }
            ObjStat stat;
            if (strlen(statInfo[i].key) >= MAX_KEY_SIZE) {
                continue;
            }
            CopyKey(stat.key, statInfo[i].key, KEY_MAX_SIZE);
            stat.size = statInfo[i].size;
            stat.time = statInfo[i].time;
            objs.insert({stat.key, stat});
        }

        FreeMemRequest freeReq = {req.comm, 1, 1, {0}};
        freeReq.addr[0] = rsp.addrOffset;
        ret = net::BioClientNet::Instance()->SendAsync<FreeMemRequest>(INVALID_NID, BIO_OP_SDK_FREE_MEM, freeReq);
        if (ret != BIO_OK) {
            CLIENT_LOG_ERROR("Send async free request failed, ret:" << ret << ".");
        }
    }
    return BIO_OK;
}

BResult BioClientAgent::SendStatRequestLocal(StatRequest &req, ObjStat &objInfo)
{
    return net::BioClientNet::Instance()->SendSync<StatRequest, ObjStat>(INVALID_NID, BIO_OP_SDK_STAT, req, objInfo);
}

BResult BioClientAgent::StatLocal(StatRequest &req, ObjStat &objInfo)
{
    if (mMode == CONVERGENCE) {
        StatResponse rsp{};
        auto ret = statOp(&req, &rsp);
        CopyKey(objInfo.key, req.key, KEY_MAX_SIZE);
        objInfo.size = rsp.size;
        objInfo.time = rsp.time;
        return ret;
    } else {
        return SendStatRequestLocal(req, objInfo);
    }
}

BResult BioClientAgent::SendNotifyUpdateRequestLocal(NotifyUpdateRequest &req)
{
    return net::BioClientNet::Instance()->SendAsync<NotifyUpdateRequest>(INVALID_NID, BIO_OP_SDK_NOTIFY_UPDATE, req);
}

BResult BioClientAgent::NotifyUpdate(bool &flag)
{
    NotifyUpdateRequest req = {{MESSAGE_MAGIC, 0, 0, mLocalNid.VNodeId(), getpid()}, flag};
    if (mMode == CONVERGENCE) {
        return notifyUpdateOp(&req);
    } else {
        return SendNotifyUpdateRequestLocal(req);
    }
}

BResult BioClientAgent::SendCheckUpdateReadyRequestLocal(CheckUpdateReadyRequest &req, CheckUpdateReadyResponse &rsp)
{
    return net::BioClientNet::Instance()->SendSync<CheckUpdateReadyRequest, CheckUpdateReadyResponse>(
        INVALID_NID, BIO_OP_SDK_CHECK_UPDATE_READY, req, rsp);
}

BResult BioClientAgent::CheckUpdateReady()
{
    BResult ret = BIO_INNER_ERR;
    CheckUpdateReadyRequest req = {{MESSAGE_MAGIC, 0, 0, mLocalNid.VNodeId(), getpid()}};
    CheckUpdateReadyResponse rsp;
    if (mMode == CONVERGENCE) {
        ret = checkUpdateReadyOp(&req, &rsp);
    } else {
        ret = SendCheckUpdateReadyRequestLocal(req, rsp);
    }
    if (ret != BIO_OK) {
        CLIENT_LOG_ERROR("Send check update ready request failed, ret:" << ret << ".");
        return ret;
    }
    return rsp.flag ? BIO_OK : BIO_ERR;
}

BResult BioClientAgent::ListLocal(ListRequest &req, std::unordered_map<std::string, ObjStat> &objs)
{
    if (mMode == CONVERGENCE) {
        return CallServerListIntf(req, objs);
    } else {
        return SendListRequestLocal(req, objs);
    }
}

BResult BioClientAgent::SendLoadRequestLocal(LoadRequest &req)
{
    BResult result = BIO_OK;
    auto ret = net::BioClientNet::Instance()->SendSync<LoadRequest, BResult>(INVALID_NID, BIO_OP_SDK_LOAD, req, result);
    if (ret != BIO_OK) {
        return ret;
    }
    return result;
}

BResult BioClientAgent::LoadLocal(LoadRequest &req)
{
    if (mMode == CONVERGENCE) {
        return loadOp(&req);
    } else {
        return SendLoadRequestLocal(req);
    }
}

BResult BioClientAgent::CalcCacheResourceLocal(CacheResourceRequest &req, std::vector<CacheResourcesDesc> &nodeDesc)
{
    CacheResourceResponse rsp;
    BResult ret = BIO_OK;
    if (mMode == CONVERGENCE) {
        ret = cacheResourceOp(&rsp);
    } else {
        ret = SendCacheResourceRequestLocal(req, rsp);
    }
    if (ret != BIO_OK) {
        CLIENT_LOG_ERROR("Send calc resource request failed, ret:" << ret << ".");
        return ret;
    }
    CacheResourcesDesc tempDesc;
    tempDesc.nodeId = rsp.nodeId;
    tempDesc.rCacheMemCapacity = rsp.rCacheMemCapacity;
    tempDesc.rCacheDiskCapacity = rsp.rCacheDiskCapacity;
    tempDesc.wCacheMemCapacity = rsp.wCacheMemCapacity;
    tempDesc.wCacheDiskCapacity = rsp.wCacheDiskCapacity;
    tempDesc.rCacheDiskUsedSize = rsp.rCacheDiskUsedSize;
    tempDesc.rCacheMemUsedSize = rsp.rCacheMemUsedSize;
    tempDesc.wCacheMemUsedSize = rsp.wCacheMemUsedSize;
    tempDesc.wCacheDiskUsedSize = rsp.wCacheDiskUsedSize;
    nodeDesc.push_back(tempDesc);
    return ret;
}

BResult BioClientAgent::SendCacheHitRequestLocal(CacheHitRequest &req, CacheHitResponse &rsp)
{
    BResult ret = net::BioClientNet::Instance()->SendSync<CacheHitRequest, CacheHitResponse>(
        INVALID_NID, BIO_OP_SDK_GET_CACHE_HIT, req, rsp);
    return ret;
}

BResult BioClientAgent::GetCacheHitLocal(CacheHitRequest &req, std::unordered_map<uint16_t, CacheHitDesc> &nodeDesc)
{
    CacheHitResponse rsp;
    BResult ret = BIO_OK;
    if (mMode == CONVERGENCE) {
        ret = cacheHitOp(&rsp);
    } else {
        ret = SendCacheHitRequestLocal(req, rsp);
    }
    if (ret != BIO_OK) {
        CLIENT_LOG_ERROR("Send get cache hit request failed, ret:" << ret << ".");
        return ret;
    }

    nodeDesc[rsp.nodeId].rCacheHitMemCount.store(rsp.rCacheHitMemCount);
    nodeDesc[rsp.nodeId].rCacheHitDiskCount.store(rsp.rCacheHitDiskCount);
    nodeDesc[rsp.nodeId].rCacheHitCount.store(rsp.rCacheHitCount);
    nodeDesc[rsp.nodeId].rCacheTotalCount.store(rsp.rCacheTotalCount);
    nodeDesc[rsp.nodeId].wCacheHitMemCount.store(rsp.wCacheHitMemCount);
    nodeDesc[rsp.nodeId].wCacheHitDiskCount.store(rsp.wCacheHitDiskCount);
    nodeDesc[rsp.nodeId].wCacheHitCount.store(rsp.wCacheHitCount);
    nodeDesc[rsp.nodeId].wCacheTotalCount.store(rsp.wCacheTotalCount);
    nodeDesc[rsp.nodeId].backendHitCount.store(rsp.backendHitCount);
    return ret;
}

BResult BioClientAgent::SendCacheResourceRequestLocal(CacheResourceRequest &req, CacheResourceResponse &rsp)
{
    BResult ret = net::BioClientNet::Instance()->SendSync<CacheResourceRequest, CacheResourceResponse>(
        INVALID_NID, BIO_OP_SDK_QUERY_CACHE_RESOURCE, req, rsp);
    return ret;
}

BResult BioClientAgent::SendGetLocalTracePointsRequest(GetTracePointsRequest &req, GetTracePointsResponse &rsp)
{
    BResult ret = net::BioClientNet::Instance()->SendSync<GetTracePointsRequest, GetTracePointsResponse>(
        INVALID_NID, BIO_OP_SDK_GET_TRACE_POINTS, req, rsp);
    return ret;
}

BResult BioClientAgent::GetTracePointsLocal(GetTracePointsRequest &req,
                                            std::map<uint16_t, TraceDatabase> &nodesTracePoints)
{
    GetTracePointsResponse rsp;
    BResult ret = BIO_OK;
    if (mMode == CONVERGENCE) {
        ret = getTracePointsOp(&rsp);
    } else {
        ret = SendGetLocalTracePointsRequest(req, rsp);
    }
    if (ret != BIO_OK) {
        CLIENT_LOG_ERROR("Send get local trace points request failed, ret: " << ret << ".");
        return ret;
    }

    if (rsp.traceDatabase.count == 0) {
        CLIENT_LOG_ERROR("Trace number is 0.");
        return BIO_ERR;
    }

    nodesTracePoints[req.nodeId] = rsp.traceDatabase;
    return BIO_OK;
}