/*
 * 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 "rcache.h"
#include <cstdint>
#include "bio_config_instance.h"
#include "bio_crc_util.h"
#include "bio_trace.h"
#include "cache_flow.h"
#include "cache_slice.h"
#include "flow_manager.h"
#include "underfs.h"

using namespace ock::bio;

static constexpr uint32_t RCACHE_FLOW_PREFIX_START = RCACHE_FLOW_MEM_META_PREFIX;

RCache::RCache(uint16_t ptId, uint64_t ptv, uint16_t diskId, uint32_t workIndex)
    : mFlowId(0),
      mPtId(ptId),
      mPtv(ptv),
      mDiskId(diskId),
      mWorkIndex(workIndex)
{
    for (int32_t tier = 0; tier < READ_CACHE_TIER_BUTT; tier++) {
        flow[tier] = nullptr;
        flow[tier] = nullptr;
        cacheData[tier] = 0ULL;
        gcData[tier] = 0ULL;
    }
}

RCache::~RCache()
{
    for (int32_t tier = 0; tier < READ_CACHE_TIER_BUTT; tier++) {
        flow[tier] = nullptr;
        flow[tier] = nullptr;
        cacheData[tier] = 0ULL;
        gcData[tier] = 0ULL;
    }
}

BResult RCache::InsertToIndex(const Key &key, RCacheChunkPtr &chunk)
{
    uint32_t bucket = GetHashBucketByKey(key);

    indexLock[bucket].Lock();
    auto iter = index[bucket].find(key);
    if (UNLIKELY(iter != index[bucket].end())) {
        index[bucket][key] = chunk;
        indexLock[bucket].UnLock();
        LOG_WARN("Repeat insert, key:" << key);
        return BIO_EXISTS;
    }

    index[bucket][key] = chunk;
    indexLock[bucket].UnLock();
    return BIO_OK;
}

BResult RCache::DeleteFromIndex(const Key &key, RCacheChunkPtr &chunk)
{
    uint32_t bucket = GetHashBucketByKey(key);

    indexLock[bucket].Lock();
    auto iter = index[bucket].find(key);
    if (iter == index[bucket].end()) {
        indexLock[bucket].UnLock();
        LOG_DEBUG("Delete read cache key:" << key << " have not exist.");
        return BIO_NOT_EXISTS;
    }
    RCacheChunkPtr ichunk = iter->second;
    if ((ichunk->GetValue().indexInFlow != chunk->GetValue().indexInFlow) ||
        (ichunk->GetValue().flowOffset != chunk->GetValue().flowOffset) ||
        (ichunk->GetValue().length != chunk->GetValue().length)) {
        indexLock[bucket].UnLock();
        LOG_DEBUG("Expired chunk, key:" << key);
        return BIO_OK;
    }
    index[bucket].erase(iter);
    indexLock[bucket].UnLock();
    return BIO_OK;
}

void RCache::AddToEvictList(RCacheTierType tierType, MqType mType, RCacheChunkPtr &chunk)
{
    evictMqLock[tierType][mType].Lock();
    evictMq[tierType][mType].PushBack(chunk);
    evictMqLock[tierType][mType].UnLock();
}

void RCache::DelFromEvictList(RCacheTierType tierType, MqType mType, RCacheChunkPtr &chunk)
{
    evictMqLock[tierType][mType].Lock();
    evictMq[tierType][mType].Remove(chunk);
    evictMqLock[tierType][mType].UnLock();
}

void RCache::AddToTruncateList(RCacheTierType tierType, RCacheChunkPtr &chunk)
{
    truncateLock[tierType].Lock();
    if (truncateQ[tierType].IsEmpty()) {
        truncateQ[tierType].PushBack(chunk);
        truncateLock[tierType].UnLock();
        return;
    }

    for (auto iter = truncateQ[tierType].Begin(); iter != nullptr;) {
        if (chunk->GetValue().indexInFlow < iter->GetValue().indexInFlow) {
            iter = iter->next[tierType];
            continue;
        }
        truncateQ[tierType].InsertAt((iter), chunk);
        truncateLock[tierType].UnLock();
        return;
    }

    truncateQ[tierType].PushBack(chunk);
    truncateLock[tierType].UnLock();
}

void RCache::DelFromTruncateList(RCacheTierType tierType, RCacheChunkPtr &chunk)
{
    truncateLock[tierType].Lock();
    truncateQ[tierType].Remove(chunk);
    truncateLock[tierType].UnLock();
}

uint32_t RCache::GetHashBucketByKey(const Key &key)
{
    return (std::hash<std::string>{}(key)) & READ_CACHE_META_HASH_BUCKET_MASK;
}

FlowType RCache::GetFlowTypeByTierType(RCacheTierType tierType)
{
    return (tierType == READ_CACHE_TIER_MEM) ? FLOW_MEMORY : FLOW_DISK;
}

BResult RCache::CreateRCacheFlow(RCacheTierType tier, std::vector<uint64_t> flowIds)
{
    flow[tier] = MakeRef<RCacheFlow>();
    if (UNLIKELY(flow[tier] == nullptr)) {
        LOG_ERROR("Alloc ptId" << mPtId << " read cache memory meta flow failed.");
        return BIO_ERR;
    }

    int32_t ret = flow[tier]->Initialize(mPtId, mDiskId, GetFlowTypeByTierType(tier), flowIds);
    if (UNLIKELY(ret != BIO_OK)) {
        LOG_ERROR("Init ptId" << mPtId << " memory meta flow failed, error code " << ret);
        Destroy();
        return BIO_ERR;
    }

    return BIO_OK;
}

BResult RCache::Initialize()
{
    std::vector<uint32_t> prefix;
    std::vector<uint64_t> flowIds;
    std::vector<uint64_t> tempIds;

    uint32_t flowPrefix;
    for (uint32_t i = RCACHE_FLOW_MEM_META_PREFIX; i <= RCACHE_FLOW_DISK_DATA_PREFIX; i++) {
        flowPrefix = CacheFlowIdManager::GenerateCacheFlowIdPrefix(mPtId, mPtv, READ_CACHE, i);
        prefix.push_back(flowPrefix);
    }

    auto flowIdAllocator = FlowIdAllocator::Instance();
    if (UNLIKELY(flowIdAllocator == nullptr)) {
        LOG_ERROR("Make flow id allocator instance failed.");
        return BIO_ALLOC_FAIL;
    }
    flowIdAllocator->GenerateFlowIds(prefix, flowIds);
    if (UNLIKELY(flowIds.empty())) {
        LOG_ERROR("Generate ptId " << mPtId << " flow ids failed.");
        return BIO_ERR;
    }

    tempIds.push_back(flowIds.at(RCACHE_FLOW_MEM_META_PREFIX - RCACHE_FLOW_PREFIX_START));
    tempIds.push_back(flowIds.at(RCACHE_FLOW_MEM_DATA_PREFIX - RCACHE_FLOW_PREFIX_START));
    auto ret = CreateRCacheFlow(READ_CACHE_TIER_MEM, tempIds);
    if (UNLIKELY(ret != BIO_OK)) {
        LOG_ERROR("Init ptId" << mPtId << " memory meta flow failed, error code " << ret);
        Destroy();
        return BIO_ERR;
    }

    tempIds.clear();
    tempIds.push_back(flowIds.at(RCACHE_FLOW_DISK_META_PREFIX - RCACHE_FLOW_PREFIX_START));
    tempIds.push_back(flowIds.at(RCACHE_FLOW_DISK_DATA_PREFIX - RCACHE_FLOW_PREFIX_START));
    ret = CreateRCacheFlow(READ_CACHE_TIER_DISK, tempIds);
    if (UNLIKELY(ret != BIO_OK)) {
        LOG_ERROR("Init ptId" << mPtId << " disk meta flow failed, error code " << ret);
        Destroy();
        return BIO_ERR;
    }

    for (auto &i : truncateQ) {
        i.Initialize(RCACHE_TRUNK_LIST_TYPE_TRUNCATE);
    }

    for (auto &i : evictMq) {
        for (auto &j : i) {
            j.Initialize(RCACHE_TRUNK_LIST_TYPE_EVICT);
        }
    }

    mFlowId = CacheFlowIdManager::GenOutFlowId(tempIds[0]);
    mCrcEnable = BioConfig::Instance()->GetDaemonConfig().enableCrc;
    return BIO_OK;
}

void RCache::Destroy()
{
    for (uint32_t i = 0; i < READ_CACHE_META_HASH_BUCKET_NUM; i++) {
        indexLock[i].Lock();
        index[i].clear();
        indexLock[i].UnLock();
    }

    for (int32_t tier = 0; tier < READ_CACHE_TIER_BUTT; tier++) {
        if (flow[tier] != nullptr) {
            flow[tier]->Destroy();
        }

        for (uint32_t i = 0; i < MQ_TYPE_BUTT; i++) {
            evictMqLock[tier][i].Lock();
            while (!evictMq[tier][i].IsEmpty()) {
                evictMq[tier][i].PopFront();
            }
            evictMqLock[tier][i].UnLock();
        }

        truncateLock[tier].Lock();
        while (!truncateQ[tier].IsEmpty()) {
            truncateQ[tier].PopFront();
        }
        truncateLock[tier].UnLock();
    }
}

BResult RCache::AllocChunk(const Key key, const RCacheValue value, RCacheChunkPtr &chunk)
{
    uint32_t keyLen = strlen(key) + 1;

    Key chunkKey = new (std::nothrow) char[keyLen];
    if (UNLIKELY(chunkKey == nullptr)) {
        LOG_ERROR("Alloc chunk key memory failed.");
        return BIO_ALLOC_FAIL;
    }

    (void)memcpy_s(chunkKey, keyLen, key, keyLen);

    chunk = MakeRef<RCacheChunk>(chunkKey, value);
    if (UNLIKELY(chunk == nullptr)) {
        delete[] chunkKey;
        LOG_ERROR("Alloc chunk for read cache key failed.");
        return BIO_ALLOC_FAIL;
    }

    return BIO_OK;
}

BResult RCache::GetSliceFromChunkIO(RCacheTierType tier, const RCacheChunkPtr &chunk, WCacheSlicePtr &slicePtr,
                                    uint64_t offset, uint64_t len, uint64_t &realLen)
{
    RCacheValue value = chunk->GetValue();
    if (UNLIKELY(offset >= value.length)) {
        LOG_ERROR("Read exceed, flow offset:" << value.flowOffset << ", length:" << value.length
                                              << ", input offset:" << offset << ", input length:" << len << ".");
        return BIO_READ_EXCEED;
    }

    std::vector<FlowAddr> flowAdd;
    realLen = value.length - offset;
    if (realLen > len) {
        realLen = len;
    }
    BResult ret = flow[tier]->GetDataFlow()->GetAddrByOffset(value.flowOffset + offset, realLen, flowAdd);
    if (UNLIKELY(ret != BIO_OK)) {
        LOG_ERROR("Get addr failed, ret:" << ret << ", key: " << chunk->GetKey() << ", tier:" << tier
                                          << ", flowOffset:" << value.flowOffset << ", length:" << value.length << ".");
        return ret;
    }

    uint64_t flowId = flow[tier]->GetDataFlow()->GetFlowId();
    FlowType flowType = GetFlowTypeByTierType(tier);
    slicePtr = MakeRef<WCacheSlice>(flowId, value.flowOffset + offset, 0ULL, realLen, flowAdd, flowType);
    if (UNLIKELY(slicePtr == nullptr)) {
        LOG_ERROR("Make slice point failed.");
        return BIO_ALLOC_FAIL;
    }

    return BIO_OK;
}

BResult RCache::GetSliceFromChunk(RCacheTierType tier, const RCacheChunkPtr &chunk, WCacheSlicePtr &slicePtr)
{
    RCacheValue value = chunk->GetValue();
    std::vector<FlowAddr> flowAdd;

    BResult ret = flow[tier]->GetDataFlow()->GetAddrByOffset(value.flowOffset, value.length, flowAdd);
    if (UNLIKELY(ret != BIO_OK)) {
        LOG_ERROR("Get key " << chunk->GetKey() << " tier " << tier << " offset " << value.flowOffset << " len "
                             << value.length << " failed.");
        return ret;
    }

    uint64_t flowId = flow[tier]->GetDataFlow()->GetFlowId();
    FlowType flowType = GetFlowTypeByTierType(tier);
    slicePtr = MakeRef<WCacheSlice>(flowId, value.flowOffset, 0ULL, value.length, flowAdd, flowType);
    if (UNLIKELY(slicePtr == nullptr)) {
        LOG_ERROR("Alloc slice ptr for read cache failed.");
        return BIO_ERR;
    }

    return BIO_OK;
}

void RCache::GetCacheResource(uint64_t &memCap, uint64_t &memUsed, uint64_t &diskCap, uint64_t &diskUsed)
{
    auto config = BioConfig::Instance()->GetDaemonConfig();
    memCap = (static_cast<uint64_t>(config.memReadRatio) * config.memCap) / NO_10;
    memUsed = FlowManager::GetCacheUsedSize(FLOW_RCACHE, FLOW_MEMORY, 0);
    diskCap = 0;
    diskUsed = 0;
    for (uint32_t diskId = 0; diskId < config.diskCaps.size(); diskId++) {
        diskCap += static_cast<uint64_t>(config.diskCaps[diskId]);
        diskUsed += FlowManager::GetCacheUsedSize(FLOW_RCACHE, FLOW_DISK, diskId);
    }
    diskCap = diskCap * static_cast<uint64_t>(config.diskReadRatio) / NO_10;
}

BResult RCache::AllocResources(uint64_t length, WCacheSlicePtr &slice)
{
    uint64_t flowId = flow[READ_CACHE_TIER_MEM]->GetDataFlow()->GetFlowId();
    uint64_t offset;
    uint64_t indexInFlow;
    std::vector<FlowAddr> flowAdd;

    BResult ret = BIO_INNER_ERR;
    BIO_TP_START(RCACHE_GET_MEM_SLICE_FAIL, &ret, BIO_INNER_RETRY);
    ret = flow[READ_CACHE_TIER_MEM]->AllocOffset(length, offset, indexInFlow);
    BIO_TP_END;
    if (ret != BIO_OK) {
        LOG_ERROR("Get tier:" << READ_CACHE_TIER_MEM << ", offset:" << offset << ", len:" << length
                              << ", flow address failed.");
        return ret;
    }
    ret = flow[READ_CACHE_TIER_MEM]->GetDataFlow()->GetAddrByOffset(offset, length, flowAdd);
    if (UNLIKELY(ret != BIO_OK)) {
        LOG_ERROR("Get tier:" << READ_CACHE_TIER_MEM << ", offset:" << offset << ", len:" << length
                              << ", flow address failed.");
        return ret;
    }

    slice = MakeRef<WCacheSlice>(flowId, offset, indexInFlow, length, flowAdd, FLOW_MEMORY);
    if (UNLIKELY(slice == nullptr)) {
        LOG_ERROR("Alloc slice memory for read cache failed.");
        return BIO_ERR;
    }

    return BIO_OK;
}

BResult RCache::Put(const Key &key, const WCacheSlicePtr &slice)
{
    RCacheValue value(slice->GetIndexInFlow(), slice->GetOffsetInFlow(), slice->GetLength());

    RCacheChunkPtr chunk = nullptr;
    BIO_TRACE_START(RCACHE_TRACE_PUT_ALLOC_CHUNK);
    auto ret = AllocChunk(key, value, chunk);
    BIO_TRACE_END(RCACHE_TRACE_PUT_ALLOC_CHUNK, ret);
    if (UNLIKELY(ret != BIO_OK)) {
        LOG_ERROR("Alloc chunk for read cache key " << key << " failed.");
        return BIO_ERR;
    }

    chunk->lock.lock();
    chunk->SetMqType(MQ_COLD);
    chunk->SetTierType(READ_CACHE_TIER_MEM);
    chunk->SetDataCrc(slice->GetDataCrc());

    LOG_DEBUG("Read cache Put, key:" << chunk->GetKey() << ", type:" << chunk->GetTierType() << ", length:"
                                     << chunk->GetValue().length << ", flowoffset:" << chunk->GetValue().flowOffset
                                     << ", indexofflow:" << chunk->GetValue().indexInFlow << ", flowId:" << mFlowId
                                     << ", crc:" << chunk->GetDataCrc());

    BIO_TRACE_START(RCACHE_TRACE_PUT_INSERT_INDEX);
    ret = InsertToIndex(chunk->GetKey(), chunk);
    BIO_TRACE_END(RCACHE_TRACE_PUT_INSERT_INDEX, ret);
    if (UNLIKELY((ret != BIO_OK) && (ret != BIO_EXISTS))) {
        chunk->lock.unlock();
        LOG_ERROR("Insert read cache key " << key << " to index failed.");
        return BIO_ERR;
    }

    BIO_TRACE_START(RCACHE_TRACE_PUT_INSERT_EVICT);
    AddToEvictList(READ_CACHE_TIER_MEM, MQ_COLD, chunk);
    BIO_TRACE_END(RCACHE_TRACE_PUT_INSERT_EVICT, 0);
    BIO_TRACE_START(RCACHE_TRACE_PUT_INSERT_TRUNC);
    AddToTruncateList(READ_CACHE_TIER_MEM, chunk);
    BIO_TRACE_END(RCACHE_TRACE_PUT_INSERT_TRUNC, 0);
    chunk->lock.unlock();

    IncCacheData(READ_CACHE_TIER_MEM, chunk->GetValue().length);
    return BIO_OK;
}

BResult RCache::Get(const Key &key, uint64_t offset, const RCacheSlicePtr &slice, const SliceWriter &sliceWriter,
                    uint64_t &realLen)
{
    uint32_t bucket = GetHashBucketByKey(key);
    WCacheSlicePtr newSlicePtr = nullptr;

    BIO_TRACE_START(RCACHE_TRACE_GET_QUERY_INDEX);
    indexLock[bucket].Lock();
    auto iter = index[bucket].find(key);
    if (iter == index[bucket].end()) {
        indexLock[bucket].UnLock();
        BIO_TRACE_END(RCACHE_TRACE_GET_QUERY_INDEX, BIO_NOT_EXISTS);
        return BIO_NOT_EXISTS;
    }
    RCacheChunkPtr chunk = iter->second;
    indexLock[bucket].UnLock();
    BIO_TRACE_END(RCACHE_TRACE_GET_QUERY_INDEX, BIO_OK);

    chunk->lock.lock();
    if (chunk->GetState() != 0) {
        LOG_ERROR("Chunk already delete, key:" << key << ".");
        chunk->lock.unlock();
        return BIO_NOT_EXISTS;
    }

    auto tier = chunk->GetTierType();
    auto ret = GetSliceFromChunkIO(tier, chunk, newSlicePtr, offset, slice->GetLength(), realLen);
    if (UNLIKELY(ret != BIO_OK || newSlicePtr == nullptr)) {
        LOG_ERROR("Read cache alloc slice failed, ret:" << ret << ", key:" << key << ", type:" << tier
                                                        << ", length:" << chunk->GetValue().length
                                                        << ", flow offset:" << chunk->GetValue().flowOffset
                                                        << ", indexofflow:" << chunk->GetValue().indexInFlow << ".");
        chunk->lock.unlock();
        return ret;
    }

    if (mCrcEnable) {
        uint32_t readCrc = 0;
        WCacheSlicePtr completeSlice = nullptr;
        ret = GetSliceFromChunk(chunk->GetTierType(), chunk, completeSlice);
        if (UNLIKELY(ret != BIO_OK || completeSlice == nullptr)) {
            LOG_ERROR("Server rcache get verify the Crc fail, read cache alloc slice failed, ret:" << ret << ", key:"
                                                                                                   << key << ".");
            return BIO_ALLOC_FAIL;
        }
        ret = completeSlice->VerifyDataCrc(chunk->GetDataCrc(), 0, completeSlice->GetLength(), nullptr);
        if (ret != BIO_OK) {
            LOG_ERROR("Server rcache get verify the Crc fail, ret:" << ret << ", key:" << key << ".");
            return ret;
        }
        ret = newSlicePtr->CalculateDataCrc(readCrc, 0, realLen);
        if (ret != BIO_OK) {
            LOG_ERROR("Server rcache get verify the CRC fail, key:" << chunk->GetKey() << ", ret: " << ret);
            return ret;
        }
        slice->SetDataCrc(readCrc);
    }

    BIO_TRACE_START(RCACHE_TRACE_GET_WRITE_DATA);
    ret = sliceWriter(newSlicePtr.Get(), slice.Get());
    BIO_TRACE_END(RCACHE_TRACE_GET_WRITE_DATA, ret);
    if (UNLIKELY(ret != BIO_OK)) {
        LOG_ERROR("Call slice writer to dst failed, ret:" << ret << ", key:" << key << ".");
        chunk->lock.unlock();
        return ret;
    }

    BIO_TRACE_START(RCACHE_TRACE_GET_UPDATE_EVICT);
    auto mqType = chunk->GetMqType();
    evictMqLock[tier][mqType].Lock();
    evictMq[tier][mqType].Remove(chunk);
    evictMq[tier][mqType].PushBack(chunk);
    evictMqLock[tier][mqType].UnLock();
    BIO_TRACE_END(RCACHE_TRACE_GET_UPDATE_EVICT, BIO_OK);
    RCacheStatistic::Instance().StatisticalByType(tier);
    chunk->lock.unlock();
    return BIO_OK;
}

BResult RCache::Load(const Key &key, uint64_t offset, uint64_t len, uint64_t &realLen)
{
    auto config = BioConfig::Instance()->GetDaemonConfig();
    auto diskCap = static_cast<uint64_t>(config.diskCaps[mDiskId]);
    uint64_t rcacheMemCap = (static_cast<uint64_t>(config.memReadRatio) * config.memCap) / NO_10;
    uint64_t rcacheMemUsed = FlowManager::GetCacheUsedSize(FLOW_RCACHE, FLOW_MEMORY, 0);
    if (rcacheMemUsed >= rcacheMemCap) {
        return BIO_ERR;
    }

    UnderFs::ObjStat stat;
    auto ret = UnderFs::Instance()->Stat(key, stat);
    if (UNLIKELY(ret != BIO_OK)) {
        LOG_ERROR("Stat key from under fs failed, ret:" << ret << ", key:" << key << ".");
        return ret;
    }

    uint64_t totalLen = static_cast<uint64_t>(stat.size);
    if (UNLIKELY(offset >= totalLen)) {
        LOG_ERROR("Read exceed, input offset:" << offset << ", input length:" << len << ", totalLen:" << totalLen);
        return BIO_READ_EXCEED;
    }
    realLen = totalLen - offset;
    if (realLen > len) {
        realLen = len;
    }

    char *value = new (std::nothrow) char[realLen];
    if (UNLIKELY(value == nullptr)) {
        LOG_ERROR("Alloc memory failed, key " << key << ", len:" << realLen << ".");
        return BIO_ALLOC_FAIL;
    }

    ret = UnderFs::Instance()->Get(key, value, realLen, offset);
    if (UNLIKELY(ret != BIO_OK)) {
        LOG_ERROR("Read data from under fs failed, ret:" << ret << ", key " << key << ", offset:" << offset
                                                         << ", length:" << realLen << ".");
        delete[] value;
        return ret;
    }

    uint64_t flowOffset = 0;
    uint64_t indexInFlow = 0;
    ret = flow[READ_CACHE_TIER_MEM]->AllocOffset(realLen, flowOffset, indexInFlow);
    if (UNLIKELY(ret != BIO_OK)) {
        LOG_ERROR("Read cache alloc offset failed, ret:" << ret << ",  key " << key << ".");
        delete[] value;
        return BIO_LOAD_ALLOC_FAIL;
    }

    RCacheValue chunkValue(indexInFlow, flowOffset, realLen);
    RCacheChunkPtr chunk = nullptr;
    ret = AllocChunk(key, chunkValue, chunk);
    if (UNLIKELY(ret != BIO_OK || chunk == nullptr)) {
        LOG_ERROR("Read cache alloc chunk failed, ret:" << ret << ",  key " << key << ".");
        delete[] value;
        return BIO_ALLOC_FAIL;
    }

    WCacheSlicePtr toSlicePtr = nullptr;
    BIO_TP_START(RCACHE_GET_MEM_SLICE_FAIL, &ret, BIO_INNER_RETRY);
    ret = GetSliceFromChunk(READ_CACHE_TIER_MEM, chunk, toSlicePtr);
    BIO_TP_END;
    if (UNLIKELY(ret != BIO_OK || toSlicePtr == nullptr)) {
        LOG_ERROR("Read cache alloc slice failed, ret:" << ret << ", key:" << key << ".");
        delete[] value;
        return BIO_ALLOC_FAIL;
    }

    ret = mSliceOperator.Copy(value, toSlicePtr.Get());
    if (UNLIKELY(ret != BIO_OK)) {
        LOG_ERROR("Copy date to slice failed, ret:" << ret << ", key " << key << " .");
        delete[] value;
        return BIO_ALLOC_FAIL;
    }

    if (mCrcEnable) {
        chunk->SetDataCrc(BioCrcUtil::Crc32(value, realLen));
    }
    chunk->lock.lock();
    chunk->SetMqType(MQ_COLD);
    chunk->SetTierType(READ_CACHE_TIER_MEM);

    delete[] value;
    ret = InsertToIndex(chunk->GetKey(), chunk);
    if (UNLIKELY(ret != BIO_OK && ret != BIO_EXISTS)) {
        chunk->lock.unlock();
        LOG_ERROR("Read cache insert index failed, ret:" << ret << ", key " << key << ".");
        return BIO_INNER_ERR;
    }

    AddToEvictList(READ_CACHE_TIER_MEM, MQ_COLD, chunk);
    AddToTruncateList(READ_CACHE_TIER_MEM, chunk);
    chunk->lock.unlock();
    IncCacheData(READ_CACHE_TIER_MEM, chunk->GetValue().length);
    return BIO_OK;
}

BResult RCache::Delete(const Key &key)
{
    uint32_t bucket = GetHashBucketByKey(key);

    BIO_TRACE_START(RCACHE_TRACE_DEL_QUERY_INDEX);
    indexLock[bucket].Lock();
    auto iter = index[bucket].find(key);
    if (UNLIKELY(iter == index[bucket].end())) {
        indexLock[bucket].UnLock();
        LOG_DEBUG("Get read cache key:" << key << " have not exist.");
        BIO_TRACE_END(RCACHE_TRACE_DEL_QUERY_INDEX, 0);
        return BIO_NOT_EXISTS;
    }
    RCacheChunkPtr chunk = iter->second;
    indexLock[bucket].UnLock();
    BIO_TRACE_END(RCACHE_TRACE_DEL_QUERY_INDEX, 0);
    chunk->lock.lock();
    chunk->SetState(1);
    chunk->lock.unlock();
    IncGCData(chunk->GetTierType(), chunk->GetValue().length);
    return BIO_OK;
}

BResult RCache::EvictMemData(const uint64_t needEvictData, uint64_t &haveEvictData)
{
    if (!mIsNormal) {
        haveEvictData = 0;
        return BIO_OK;
    }

    bool expectValue = false;
    if (!mMemEvict.compare_exchange_weak(expectValue, true)) {
        haveEvictData = 0ULL;
        return BIO_OK;
    }

    BResult ret = EvictMemDataImpl(needEvictData, haveEvictData);
    mMemEvict.store(false);
    return ret;
}

BResult RCache::EvictDiskData(const uint64_t needEvictData, uint64_t &haveEvictData)
{
    if (!mIsNormal) {
        haveEvictData = 0;
        return BIO_OK;
    }

    bool expectValue = false;
    if (!mDiskEvict.compare_exchange_weak(expectValue, true)) {
        haveEvictData = 0ULL;
        return BIO_OK;
    }

    BResult ret = EvictDiskDataImpl(needEvictData, haveEvictData);
    mDiskEvict.store(false);
    return ret;
}

bool RCache::IsEmptyEvict()
{
    if (mMemEvict) {
        LOG_DEBUG("Mem: task:" << mMemEvict);
        return false;
    }

    if (mDiskEvict) {
        LOG_DEBUG("Disk: task:" << mDiskEvict);
        return false;
    }

    return true;
}

BResult RCache::EvictMemDataImpl(const uint64_t needEvictData, uint64_t &haveEvictData)
{
    haveEvictData = 0ULL;
    RCacheChunkPtr chunk;
    RCacheChunkPtr newChunk;
    WCacheSlicePtr fromSlicePtr = nullptr;
    WCacheSlicePtr toSlicePtr = nullptr;

    while (haveEvictData < needEvictData) {
        if (!mIsNormal) {
            return BIO_OK;
        }
        truncateLock[READ_CACHE_TIER_MEM].Lock();
        if (truncateQ[READ_CACHE_TIER_MEM].IsEmpty()) {
            truncateLock[READ_CACHE_TIER_MEM].UnLock();
            break;
        }
        chunk = truncateQ[READ_CACHE_TIER_MEM].End();

        uint64_t truncateOffset = 0;
        BIO_TP_START(RCACHE_GET_EVICT_IO_FAIL, &truncateOffset, NO_MAX_VALUE64);
        truncateOffset = flow[READ_CACHE_TIER_MEM]->GetDataTruncOffset();
        BIO_TP_END;
        if (chunk->GetValue().flowOffset != truncateOffset) {
            truncateLock[READ_CACHE_TIER_MEM].UnLock();
            LOG_WARN("RCache evict stuck, need truncate offset:" << truncateOffset << ", the chunk "
                                                                 << chunk->ToString());
            break;
        }

        if (chunk->GetValue().length + haveEvictData > needEvictData) {
            truncateLock[READ_CACHE_TIER_MEM].UnLock();
            break;
        }

        truncateLock[READ_CACHE_TIER_MEM].UnLock();
        BIO_TRACE_START(RCACHE_TRACE_EVICT2DISK);
        chunk->lock.lock();
        uint64_t flowOffset = 0;
        uint64_t indexInFlow = 0;
        BResult ret = BIO_INNER_ERR;
        BIO_TP_START(RCACHE_GET_DISK_SLICE_FAIL, &ret, BIO_INNER_RETRY);
        ret = flow[READ_CACHE_TIER_DISK]->AllocOffset(chunk->GetValue().length, flowOffset, indexInFlow);
        BIO_TP_END;
        if (UNLIKELY(ret != BIO_OK)) {
            LOG_ERROR("Alloc offset for read cache key " << chunk->GetKey() << " failed.");
            chunk->lock.unlock();
            BIO_TRACE_END(RCACHE_TRACE_EVICT2DISK, ret);
            return BIO_ALLOC_FAIL;
        }
        RCacheValue chunkValue(indexInFlow, flowOffset, chunk->GetValue().length);
        ret = AllocChunk(chunk->GetKey(), chunkValue, newChunk);
        if (UNLIKELY(ret != BIO_OK) || (chunk == nullptr)) {
            LOG_ERROR("Alloc chunk for read cache key " << chunk->GetKey() << " failed.");
            chunk->lock.unlock();
            BIO_TRACE_END(RCACHE_TRACE_EVICT2DISK, ret);
            return BIO_ALLOC_FAIL;
        }
        ret = GetSliceFromChunk(READ_CACHE_TIER_MEM, chunk, fromSlicePtr);
        if (UNLIKELY(ret != BIO_OK) || (fromSlicePtr == nullptr)) {
            LOG_ERROR("RCache alloc mem tier slice failed, " << chunk->ToString());
            chunk->lock.unlock();
            BIO_TRACE_END(RCACHE_TRACE_EVICT2DISK, ret);
            return BIO_ALLOC_FAIL;
        }
        ret = GetSliceFromChunk(READ_CACHE_TIER_DISK, newChunk, toSlicePtr);
        if (UNLIKELY(ret != BIO_OK) || (toSlicePtr == nullptr)) {
            LOG_ERROR("RCache alloc disk tier slice failed,  " << newChunk->ToString());
            chunk->lock.unlock();
            BIO_TRACE_END(RCACHE_TRACE_EVICT2DISK, ret);
            return BIO_ALLOC_FAIL;
        }
        ret = mSliceOperator.Copy(fromSlicePtr.Get(), toSlicePtr.Get());
        if (UNLIKELY(ret != BIO_OK)) {
            LOG_ERROR("RCache copy mem tier slice to disk tier slice failed, " << chunk->ToString());
            chunk->lock.unlock();
            BIO_TRACE_END(RCACHE_TRACE_EVICT2DISK, ret);
            return BIO_ALLOC_FAIL;
        }
        newChunk->SetDataCrc(chunk->GetDataCrc());
        DelFromTruncateList(READ_CACHE_TIER_MEM, chunk);
        DelFromEvictList(READ_CACHE_TIER_MEM, chunk.Get()->GetMqType(), chunk);
        chunk->SetValue(newChunk->GetValue());
        chunk->SetTierType(READ_CACHE_TIER_DISK);
        AddToEvictList(READ_CACHE_TIER_DISK, MQ_COLD, chunk);
        AddToTruncateList(READ_CACHE_TIER_DISK, chunk);
        flow[READ_CACHE_TIER_MEM]->UpdateDataTruncOffset(chunk->GetValue().flowOffset, chunk->GetValue().length);
        haveEvictData += chunk->GetValue().length;
        LOG_DEBUG("RCache evict chunk to disk tier success, " << chunk->ToString());
        delete[] newChunk->GetKey();
        chunk->lock.unlock();
        BIO_TRACE_END(RCACHE_TRACE_EVICT2DISK, BIO_OK);
    }

    if (haveEvictData == 0) {
        return BIO_OK;
    }

    uint64_t truncateOffset = flow[READ_CACHE_TIER_MEM]->GetDataTruncOffset();
    auto ret = flow[READ_CACHE_TIER_MEM]->GetDataFlow()->TruncateOffset(truncateOffset);
    if (UNLIKELY(ret != BIO_OK)) {
        LOG_ERROR("Truncate read cache key " << chunk->GetKey() << " mem data flow to " << truncateOffset << " failed."
                                             << ret);
        return BIO_ALLOC_FAIL;
    }

    DecCacheData(READ_CACHE_TIER_MEM, haveEvictData);
    IncCacheData(READ_CACHE_TIER_DISK, haveEvictData);
    return BIO_OK;
}

BResult RCache::EvictDiskDataImpl(const uint64_t needEvictData, uint64_t &haveEvictData)
{
    haveEvictData = 0ULL;
    RCacheChunkPtr chunk;
    while (haveEvictData < needEvictData) {
        if (!mIsNormal) {
            return BIO_OK;
        }
        truncateLock[READ_CACHE_TIER_DISK].Lock();
        if (truncateQ[READ_CACHE_TIER_DISK].IsEmpty()) {
            truncateLock[READ_CACHE_TIER_DISK].UnLock();
            break;
        }
        chunk = truncateQ[READ_CACHE_TIER_DISK].End();
        uint64_t truncateOffset = flow[READ_CACHE_TIER_DISK]->GetDataTruncOffset();
        if (chunk->GetValue().flowOffset != truncateOffset) {
            truncateLock[READ_CACHE_TIER_DISK].UnLock();
            LOG_WARN("RCache evict stuck, need truncate offset:" << truncateOffset << ", the chunk "
                                                                 << chunk->ToString());
            break;
        }

        if (chunk->GetValue().length + haveEvictData > needEvictData) {
            truncateLock[READ_CACHE_TIER_DISK].UnLock();
            break;
        }

        truncateLock[READ_CACHE_TIER_DISK].UnLock();
        BIO_TRACE_START(RCACHE_TRACE_EVICT2NULL);
        chunk->lock.lock();
        chunk->SetState(1);
        auto ret = DeleteFromIndex(chunk->GetKey(), chunk);
        if (UNLIKELY(ret != BIO_OK)) {
            LOG_DEBUG("Get read cache key:" << chunk->GetKey() << " not exist.");
        }

        DelFromTruncateList(READ_CACHE_TIER_DISK, chunk);
        DelFromEvictList(chunk->GetTierType(), chunk.Get()->GetMqType(), chunk);
        flow[READ_CACHE_TIER_DISK]->UpdateDataTruncOffset(chunk->GetValue().flowOffset, chunk->GetValue().length);
        haveEvictData += chunk->GetValue().length;
        LOG_DEBUG("Delete chunk, key: " << chunk->GetKey() << ", type:" << chunk->GetTierType() << ", length:"
                                        << chunk->GetValue().length << ", flowOffset:" << chunk->GetValue().flowOffset
                                        << ", indexInFlow:" << chunk->GetValue().indexInFlow);
        chunk->lock.unlock();

        delete[] chunk->GetKey();
        BIO_TRACE_END(RCACHE_TRACE_EVICT2NULL, ret);
    }

    if (haveEvictData == 0) {
        return BIO_OK;
    }

    uint64_t truncateOffset = flow[READ_CACHE_TIER_DISK]->GetDataTruncOffset();
    auto ret = flow[READ_CACHE_TIER_DISK]->GetDataFlow()->TruncateOffset(truncateOffset);
    if (UNLIKELY(ret != BIO_OK)) {
        LOG_ERROR("Truncate disk data to offset " << truncateOffset << " flow failed." << ret);
        return BIO_ALLOC_FAIL;
    }

    DecCacheData(READ_CACHE_TIER_DISK, haveEvictData);
    return BIO_OK;
}