#4999

Committed 22 Mar 2026 10:21AM UTC coverage: 72.31% (-0.03%) from 72.335%

Build # #4999

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

feat(subq/some): some/any/exists for stream subq (#34860)

Coverage Stats

50 of 68 new or added lines in 1 file covered. (73.53%)

701 existing lines in 132 files now uncovered.

253472 of 350536 relevant lines covered (72.31%)

131775367.32 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

69.25

/source/util/src/tpagedbuf.c

#define _DEFAULT_SOURCE
#include "tpagedbuf.h"
#include "taoserror.h"
#include "tcompression.h"
#include "tlog.h"
#include "tsimplehash.h"

#define GET_PAYLOAD_DATA(_p)           ((char*)(_p)->pData + POINTER_BYTES)
#define BUF_PAGE_IN_MEM(_p)            ((_p)->pData != NULL)
#define CLEAR_BUF_PAGE_IN_MEM_FLAG(_p) ((_p)->pData = NULL)
#define HAS_DATA_IN_DISK(_p)           ((_p)->offset >= 0)
#define NO_IN_MEM_AVAILABLE_PAGES(_b)  (listNEles((_b)->lruList) >= (_b)->inMemPages)

typedef struct SPageDiskInfo {
  int64_t offset;
  int32_t length;
} SPageDiskInfo, SFreeListItem;

struct SPageInfo {
  SListNode* pn;  // point to list node struct. it is NULL when the page is evicted from the in-memory buffer
  void*      pData;
  int64_t    offset;
  int32_t    pageId;
  int32_t    length : 29;
  bool       used : 1;   // set current page is in used
  bool       dirty : 1;  // set current buffer page is dirty or not
};

struct SDiskbasedBuf {
  int32_t    numOfPages;
  int64_t    totalBufSize;
  uint64_t   fileSize;  // disk file size
  TdFilePtr  pFile;
  int32_t    allocateId;  // allocated page id
  char*      path;        // file path
  char*      prefix;      // file name prefix
  int32_t    pageSize;    // current used page size
  int32_t    inMemPages;  // numOfPages that are allocated in memory
  SList*     freePgList;  // free page list
  SArray*    pIdList;     // page id list
  SSHashObj* all;
  SList*     lruList;
  void*      emptyDummyIdList;  // dummy id list
  void*      assistBuf;         // assistant buffer for compress/decompress data
  SArray*    pFree;             // free area in file
  bool       comp;              // compressed before flushed to disk
  uint64_t   nextPos;           // next page flush position

  char*               id;           // for debug purpose
  bool                printStatis;  // Print statistics info when closing this buffer.
  SDiskbasedBufStatis statis;
};

static int32_t createDiskFile(SDiskbasedBuf* pBuf) {
  if (pBuf->path == NULL) {  // prepare the file name when needed it
    char path[PATH_MAX] = {0};
    taosGetTmpfilePath(pBuf->prefix, "paged-buf", path);
    pBuf->path = taosStrdup(path);
    if (pBuf->path == NULL) {
      return terrno;
    }
  }

  pBuf->pFile =
      taosOpenFile(pBuf->path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_TRUNC | TD_FILE_AUTO_DEL);
  if (pBuf->pFile == NULL) {
    return terrno;
  }

  int64_t realSize = -1;
  if (taosFStatFile(pBuf->pFile, &realSize, NULL) != TSDB_CODE_SUCCESS) {
    realSize = -1;
  }
  uDebug("paged buffer file opened, path:%s, realSize:%" PRId64 ", nextPos:%" PRIu64 ", fileSize:%" PRIu64 ", %s",
         pBuf->path, realSize, pBuf->nextPos, pBuf->fileSize, pBuf->id);

  return TSDB_CODE_SUCCESS;
}

static char* doCompressData(void* data, int32_t srcSize, int32_t* dst, SDiskbasedBuf* pBuf) {  // do nothing
  if (!pBuf->comp) {
    *dst = srcSize;
    return data;
  }

  *dst = tsCompressString(data, srcSize, 1, pBuf->assistBuf, srcSize, ONE_STAGE_COMP, NULL, 0);

  memcpy(data, pBuf->assistBuf, *dst);
  return data;
}

static int32_t doDecompressData(void* data, int32_t srcSize, int32_t* dst, SDiskbasedBuf* pBuf) {  // do nothing
  int32_t code = 0;
  if (!pBuf->comp) {
    *dst = srcSize;
    return code;
  }

  *dst = tsDecompressString(data, srcSize, 1, pBuf->assistBuf, pBuf->pageSize, ONE_STAGE_COMP, NULL, 0);
  if (*dst > 0) {
    memcpy(data, pBuf->assistBuf, *dst);
  } else if (*dst < 0) {
    return terrno;
  }
  return code;
  ;
}

static uint64_t allocateNewPositionInFile(SDiskbasedBuf* pBuf, size_t size) {
  if (pBuf->pFree == NULL) {
    return pBuf->nextPos;
  } else {
    int32_t offset = -1;

    size_t num = taosArrayGetSize(pBuf->pFree);
    for (int32_t i = 0; i < num; ++i) {
      SFreeListItem* pi = taosArrayGet(pBuf->pFree, i);
      if (pi->length >= size) {
        offset = pi->offset;
        pi->offset += (int32_t)size;
        pi->length -= (int32_t)size;

        return offset;
      }
    }

    // no available recycle space, allocate new area in file
    return pBuf->nextPos;
  }
}

/**
 *   +--------------------------+-------------------+--------------+
 *   | PTR to SPageInfo (8bytes)| Payload (PageSize)| 2 Extra Bytes|
 *   +--------------------------+-------------------+--------------+
 * @param pBuf
 * @param pg
 * @return
 */

static FORCE_INLINE size_t getAllocPageSize(int32_t pageSize) { return pageSize + POINTER_BYTES + sizeof(SFilePage); }

static int32_t doFlushBufPageImpl(SDiskbasedBuf* pBuf, int64_t offset, const char* pData, int32_t size) {
  int64_t ret = taosLSeekFile(pBuf->pFile, offset, SEEK_SET);
  if (ret < 0) {
    return terrno;
  }

  ret = (int32_t)taosWriteFile(pBuf->pFile, pData, size);
  if (ret != size) {
    return terrno;
  }

  // extend the file
  if (pBuf->fileSize < offset + size) {
    pBuf->fileSize = offset + size;
  }

  pBuf->statis.flushBytes += size;
  pBuf->statis.flushPages += 1;

  return TSDB_CODE_SUCCESS;
}

static char* doFlushBufPage(SDiskbasedBuf* pBuf, SPageInfo* pg) {
  if (pg->pData == NULL || pg->used) {
    uError("invalid params in paged buffer process when flushing buf to disk, %s", pBuf->id);
    terrno = TSDB_CODE_INVALID_PARA;
    return NULL;
  }

  int32_t size = pBuf->pageSize;
  int64_t offset = pg->offset;

  char* t = NULL;
  if ((!HAS_DATA_IN_DISK(pg)) || pg->dirty) {
    void* payload = GET_PAYLOAD_DATA(pg);
    t = doCompressData(payload, pBuf->pageSize + sizeof(SFilePage), &size, pBuf);
    if (size < 0) {
      uError("failed to compress data when flushing data to disk, %s", pBuf->id);
      terrno = TSDB_CODE_INVALID_PARA;
      return NULL;
    }
  }

  // this page is flushed to disk for the first time
  if (pg->dirty) {
    if (!HAS_DATA_IN_DISK(pg)) {
      offset = allocateNewPositionInFile(pBuf, size);
      pBuf->nextPos += size;

      int32_t code = doFlushBufPageImpl(pBuf, offset, t, size);
      if (code != TSDB_CODE_SUCCESS) {
        return NULL;
      }
    } else {
      // length becomes greater, current space is not enough, allocate new place, otherwise, do nothing
      if (pg->length < size) {
        // 1. add current space to free list
        SPageDiskInfo dinfo = {.length = pg->length, .offset = offset};
        if (NULL == taosArrayPush(pBuf->pFree, &dinfo)) {
          return NULL;
        }

        // 2. allocate new position, and update the info
        offset = allocateNewPositionInFile(pBuf, size);
        pBuf->nextPos += size;
      }

      int32_t code = doFlushBufPageImpl(pBuf, offset, t, size);
      if (code != TSDB_CODE_SUCCESS) {
        return NULL;
      }
    }
  } else {  // NOTE: the size may be -1, the this recycle page has not been flushed to disk yet.
    size = pg->length;
  }

  char* pDataBuf = pg->pData;
  memset(pDataBuf, 0, getAllocPageSize(pBuf->pageSize));

#ifdef BUF_PAGE_DEBUG
  uDebug("page_flush %p, pageId:%d, offset:%d", pDataBuf, pg->pageId, offset);
#endif

  pg->offset = offset;
  pg->length = size;  // on disk size
  return pDataBuf;
}

static char* flushBufPage(SDiskbasedBuf* pBuf, SPageInfo* pg) {
  int32_t ret = TSDB_CODE_SUCCESS;

  if (pBuf->pFile == NULL) {
    if ((ret = createDiskFile(pBuf)) != TSDB_CODE_SUCCESS) {
      terrno = ret;
      return NULL;
    }
  }

  char* p = doFlushBufPage(pBuf, pg);
  CLEAR_BUF_PAGE_IN_MEM_FLAG(pg);

  pg->dirty = false;
  return p;
}

// load file block data in disk
static int32_t loadPageFromDisk(SDiskbasedBuf* pBuf, SPageInfo* pg) {
  if (pg->offset < 0 || pg->length <= 0) {
    uError("failed to load buf page from disk, offset:%" PRId64 ", length:%d, %s", pg->offset, pg->length, pBuf->id);
    return TSDB_CODE_INVALID_PARA;
  }

  int64_t ret = taosLSeekFile(pBuf->pFile, pg->offset, SEEK_SET);
  if (ret < 0) {
    ret = terrno;
    return ret;
  }

  void* pPage = (void*)GET_PAYLOAD_DATA(pg);
  ret = taosReadFile(pBuf->pFile, pPage, pg->length);
  if (ret != pg->length) {
    ret = terrno;
    return ret;
  }

  pBuf->statis.loadBytes += pg->length;
  pBuf->statis.loadPages += 1;

  int32_t fullSize = 0;
  return doDecompressData(pPage, pg->length, &fullSize, pBuf);
}

static SPageInfo* registerNewPageInfo(SDiskbasedBuf* pBuf, int32_t pageId) {
  pBuf->numOfPages += 1;

  SPageInfo* ppi = taosMemoryMalloc(sizeof(SPageInfo));
  if (ppi == NULL) {
    return NULL;
  }

  ppi->pageId = pageId;
  ppi->pData = NULL;
  ppi->offset = -1;
  ppi->length = -1;
  ppi->used = true;
  ppi->pn = NULL;
  ppi->dirty = false;

  SPageInfo** pRet = taosArrayPush(pBuf->pIdList, &ppi);
  if (NULL == pRet) {
    taosMemoryFree(ppi);
    return NULL;
  }
  return *pRet;
}

static SListNode* getEldestUnrefedPage(SDiskbasedBuf* pBuf) {
  SListIter iter = {0};
  tdListInitIter(pBuf->lruList, &iter, TD_LIST_BACKWARD);

  SListNode* pn = NULL;
  while ((pn = tdListNext(&iter)) != NULL) {
    SPageInfo* pageInfo = *(SPageInfo**)pn->data;

    SPageInfo* p = *(SPageInfo**)(pageInfo->pData);

    if (!pageInfo->used) {
      break;
    }
  }

  return pn;
}

static char* evictBufPage(SDiskbasedBuf* pBuf) {
  SListNode* pn = getEldestUnrefedPage(pBuf);
  if (pn == NULL) {  // no available buffer pages now, return.
    return NULL;
  }

  terrno = 0;
  pn = tdListPopNode(pBuf->lruList, pn);

  SPageInfo* d = *(SPageInfo**)pn->data;

  d->pn = NULL;
  taosMemoryFreeClear(pn);

  return flushBufPage(pBuf, d);
}

static int32_t lruListPushFront(SList* pList, SPageInfo* pi) {
  int32_t code = tdListPrepend(pList, &pi);
  if (TSDB_CODE_SUCCESS != code) {
    return code;
  }
  SListNode* front = tdListGetHead(pList);
  pi->pn = front;
  return TSDB_CODE_SUCCESS;
}

static void lruListMoveToFront(SList* pList, SPageInfo* pi) {
  pi->pn = tdListPopNode(pList, pi->pn);
  tdListPrependNode(pList, pi->pn);
}

static SPageInfo* getPageInfoFromPayload(void* page) {
  char* p = (char*)page - POINTER_BYTES;

  SPageInfo* ppi = ((SPageInfo**)p)[0];
  return ppi;
}

int32_t createDiskbasedBuf(SDiskbasedBuf** pBuf, int32_t pagesize, int64_t inMemBufSize, const char* id,
                           const char* dir) {
  int32_t code = 0;
  *pBuf = NULL;
  SDiskbasedBuf* pPBuf = taosMemoryCalloc(1, sizeof(SDiskbasedBuf));
  if (pPBuf == NULL) {
    code = terrno;
    goto _error;
  }

  pPBuf->pageSize = pagesize;
  pPBuf->numOfPages = 0;  // all pages are in buffer in the first place
  pPBuf->totalBufSize = 0;
  pPBuf->allocateId = -1;
  pPBuf->pFile = NULL;
  pPBuf->id = taosStrdup(id);
  if (id != NULL && pPBuf->id == NULL) {
    code = terrno;
    goto _error;
  }
  pPBuf->fileSize = 0;
  pPBuf->pFree = taosArrayInit(4, sizeof(SFreeListItem));
  pPBuf->freePgList = tdListNew(POINTER_BYTES);
  if (pPBuf->pFree == NULL || pPBuf->freePgList == NULL) {
    code = terrno;
    goto _error;
  }

  // at least more than 2 pages must be in memory
  if (inMemBufSize < pagesize * 2) {
    inMemBufSize = pagesize * 2;
  }

  pPBuf->inMemPages = inMemBufSize / pagesize;  // maximum allowed pages, it is a soft limit.
  pPBuf->lruList = tdListNew(POINTER_BYTES);
  if (pPBuf->lruList == NULL) {
    code = terrno;
    goto _error;
  }

  // init id hash table
  _hash_fn_t fn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT);
  pPBuf->pIdList = taosArrayInit(4, POINTER_BYTES);
  if (pPBuf->pIdList == NULL) {
    code = terrno;
    goto _error;
  }

  pPBuf->all = tSimpleHashInit(64, fn);
  if (pPBuf->all == NULL) {
    code = terrno;
    goto _error;
  }

  pPBuf->prefix = (char*)dir;
  pPBuf->emptyDummyIdList = taosArrayInit(1, sizeof(int32_t));
  if (pPBuf->emptyDummyIdList == NULL) {
    code = terrno;
    goto _error;
  }

  //  qDebug("QInfo:0x%"PRIx64 ", create resBuf for output, page size:%d, inmem buf pages:%d, file:%s", qId,
  //  pPBuf->pageSize, pPBuf->inMemPages, pPBuf->path);

  *pBuf = pPBuf;
  return TSDB_CODE_SUCCESS;

_error:
  destroyDiskbasedBuf(pPBuf);
  *pBuf = NULL;
  return code;
}

static char* doExtractPage(SDiskbasedBuf* pBuf) {
  char* availablePage = NULL;
  if (NO_IN_MEM_AVAILABLE_PAGES(pBuf)) {
    availablePage = evictBufPage(pBuf);
    if (availablePage == NULL) {
      uWarn("no available buf pages, current:%d, max:%d, reason: %s, %s", listNEles(pBuf->lruList), pBuf->inMemPages,
            terrstr(), pBuf->id)
    }
  } else {
    availablePage =
        taosMemoryCalloc(1, getAllocPageSize(pBuf->pageSize));  // add extract bytes in case of zipped buffer increased.
  }

  return availablePage;
}

void* getNewBufPage(SDiskbasedBuf* pBuf, int32_t* pageId) {
  pBuf->statis.getPages += 1;

  char* availablePage = doExtractPage(pBuf);
  if (availablePage == NULL) {
    return NULL;
  }

  SPageInfo* pi = NULL;
  int32_t    code = 0;
  if (listNEles(pBuf->freePgList) != 0) {
    SListNode* pItem = tdListPopHead(pBuf->freePgList);
    pi = *(SPageInfo**)pItem->data;
    pi->used = true;
    *pageId = pi->pageId;
    taosMemoryFreeClear(pItem);
    code = lruListPushFront(pBuf->lruList, pi);
    if (TSDB_CODE_SUCCESS != code) {
      taosMemoryFree(pi);
      taosMemoryFree(availablePage);
      terrno = code;
      return NULL;
    }
  } else {  // create a new pageinfo
    // register new id in this group
    *pageId = (++pBuf->allocateId);

    // register page id info
    pi = registerNewPageInfo(pBuf, *pageId);
    if (pi == NULL) {
      taosMemoryFree(availablePage);
      return NULL;
    }

    // add to hash map
    int32_t code = tSimpleHashPut(pBuf->all, pageId, sizeof(int32_t), &pi, POINTER_BYTES);

    if (TSDB_CODE_SUCCESS == code) {
      // add to LRU list
      code = lruListPushFront(pBuf->lruList, pi);
    }
    if (TSDB_CODE_SUCCESS == code) {
      pBuf->totalBufSize += pBuf->pageSize;
    } else {
      taosMemoryFree(availablePage);
      SPageInfo **pLast = taosArrayPop(pBuf->pIdList);
      int32_t ret = tSimpleHashRemove(pBuf->all, pageId, sizeof(int32_t));
      if (ret != TSDB_CODE_SUCCESS) {
        uError("%s failed to clear pageId %d from buf hash-set since %s", __func__, *pageId, tstrerror(ret));
      }
      taosMemoryFree(pi);
      terrno = code;
      return NULL;
    }
  }

  pi->pData = availablePage;

  ((void**)pi->pData)[0] = pi;
#ifdef BUF_PAGE_DEBUG
  uDebug("page_getNewBufPage , pi->pData:%p, pageId:%d, offset:%" PRId64, pi->pData, pi->pageId, pi->offset);
#endif

  return (void*)(GET_PAYLOAD_DATA(pi));
}

void* getBufPage(SDiskbasedBuf* pBuf, int32_t id) {
  if (id < 0) {
    terrno = TSDB_CODE_INVALID_PARA;
    uError("invalid page id:%d, %s", id, pBuf->id);
    return NULL;
  }

  pBuf->statis.getPages += 1;

  SPageInfo** pi = tSimpleHashGet(pBuf->all, &id, sizeof(int32_t));
  if (pi == NULL || *pi == NULL) {
    uError("failed to locate the buffer page:%d, %s", id, pBuf->id);
    terrno = TSDB_CODE_INVALID_PARA;
    return NULL;
  }

  if (BUF_PAGE_IN_MEM(*pi)) {  // it is in memory
    // no need to update the LRU list if only one page exists
    if (pBuf->numOfPages == 1) {
      (*pi)->used = true;
      return (void*)(GET_PAYLOAD_DATA(*pi));
    }

    SPageInfo** pInfo = (SPageInfo**)((*pi)->pn->data);
    if (*pInfo != *pi) {
      terrno = TSDB_CODE_APP_ERROR;
      uError("inconsistently data in paged buffer, pInfo:%p, pi:%p, %s", *pInfo, *pi, pBuf->id);
      return NULL;
    }

    lruListMoveToFront(pBuf->lruList, (*pi));
    (*pi)->used = true;

#ifdef BUF_PAGE_DEBUG
    uDebug("page_getBufPage1 pageId:%d, offset:%" PRId64, (*pi)->pageId, (*pi)->offset);
#endif
    return (void*)(GET_PAYLOAD_DATA(*pi));
  } else {  // not in memory

    (*pi)->pData = doExtractPage(pBuf);

    // failed to evict buffer page, return with error code.
    if ((*pi)->pData == NULL) {
      return NULL;
    }

    // set the ptr to the new SPageInfo
    ((void**)((*pi)->pData))[0] = (*pi);

    int32_t code = lruListPushFront(pBuf->lruList, *pi);
    if (TSDB_CODE_SUCCESS != code) {
      taosMemoryFree((*pi)->pData);
      (*pi)->pData = NULL;
      terrno = code;
      return NULL;
    }
    (*pi)->used = true;

    // some data has been flushed to disk, and needs to be loaded into buffer again.
    if (HAS_DATA_IN_DISK(*pi)) {
      int32_t code = loadPageFromDisk(pBuf, *pi);
      if (code != 0) {
        taosMemoryFree((*pi)->pData);
        (*pi)->pData = NULL;
        terrno = code;
        return NULL;
      }
    }
#ifdef BUF_PAGE_DEBUG
    uDebug("page_getBufPage2 pageId:%d, offset:%" PRId64, (*pi)->pageId, (*pi)->offset);
#endif
    return (void*)(GET_PAYLOAD_DATA(*pi));
  }
}

void releaseBufPage(SDiskbasedBuf* pBuf, void* page) {
  if (page == NULL) {
    return;
  }

  SPageInfo* ppi = getPageInfoFromPayload(page);
  releaseBufPageInfo(pBuf, ppi);
}

void releaseBufPageInfo(SDiskbasedBuf* pBuf, SPageInfo* pi) {
#ifdef BUF_PAGE_DEBUG
  uDebug("page_releaseBufPageInfo pageId:%d, used:%d, offset:%" PRId64, pi->pageId, pi->used, pi->offset);
#endif

  if (pi == NULL) {
    return;
  }

  if (pi->pData == NULL) {
    uError("pi->pData (page data) is null");
    return;
  }

  pi->used = false;
  pBuf->statis.releasePages += 1;
}

size_t getTotalBufSize(const SDiskbasedBuf* pBuf) { return (size_t)pBuf->totalBufSize; }

SArray* getDataBufPagesIdList(SDiskbasedBuf* pBuf) { return pBuf->pIdList; }

void destroyDiskbasedBuf(SDiskbasedBuf* pBuf) {
  if (pBuf == NULL) {
    return;
  }

  dBufPrintStatis(pBuf);

  bool needRemoveFile = false;
  if (pBuf->pFile != NULL) {
    needRemoveFile = true;
    uDebug(
        "Paged buffer closed, total:%.2f Kb (%d Pages), inmem size:%.2f Kb (%d Pages), file size:%.2f Kb, page "
        "size:%.2f Kb, %s",
        pBuf->totalBufSize / 1024.0, pBuf->numOfPages, listNEles(pBuf->lruList) * pBuf->pageSize / 1024.0,
        listNEles(pBuf->lruList), pBuf->fileSize / 1024.0, pBuf->pageSize / 1024.0f, pBuf->id);

    int32_t code = taosCloseFile(&pBuf->pFile);
    if (TSDB_CODE_SUCCESS != code) {
      uError("failed to close paged buffer file when destroying, path:%s, closeCode:%d, err:%s, %s", pBuf->path, code,
             tstrerror(code), pBuf->id);
    }
  } else {
    uDebug("Paged buffer closed, total:%.2f Kb, no file created, %s", pBuf->totalBufSize / 1024.0, pBuf->id);
  }

  // print the statistics information
  {
    SDiskbasedBufStatis* ps = &pBuf->statis;
    if (ps->loadPages == 0) {
      uDebug("Get/Release pages:%d/%d, flushToDisk:%.2f Kb (%d Pages), loadFromDisk:%.2f Kb (%d Pages)", ps->getPages,
             ps->releasePages, ps->flushBytes / 1024.0f, ps->flushPages, ps->loadBytes / 1024.0f, ps->loadPages);
    } else {
      uDebug(
          "Get/Release pages:%d/%d, flushToDisk:%.2f Kb (%d Pages), loadFromDisk:%.2f Kb (%d Pages), avgPgSize:%.2f Kb",
          ps->getPages, ps->releasePages, ps->flushBytes / 1024.0f, ps->flushPages, ps->loadBytes / 1024.0f,
          ps->loadPages, ps->loadBytes / (1024.0 * ps->loadPages));
    }
  }

  if (needRemoveFile) {
    int32_t ret = taosRemoveFile(pBuf->path);
    if (ret != 0) {  // print the error and discard this error info
      uDebug("WARNING tPage remove file failed. path=%s, code:%s", pBuf->path, strerror(ERRNO));
    }
  }

  taosMemoryFreeClear(pBuf->path);

  size_t n = taosArrayGetSize(pBuf->pIdList);
  for (int32_t i = 0; i < n; ++i) {
    SPageInfo* pi = taosArrayGetP(pBuf->pIdList, i);
    taosMemoryFreeClear(pi->pData);
    taosMemoryFreeClear(pi);
  }

  taosArrayDestroy(pBuf->pIdList);

  pBuf->lruList = tdListFree(pBuf->lruList);
  pBuf->freePgList = tdListFree(pBuf->freePgList);

  taosArrayDestroy(pBuf->emptyDummyIdList);
  taosArrayDestroy(pBuf->pFree);

  tSimpleHashCleanup(pBuf->all);

  taosMemoryFreeClear(pBuf->id);
  taosMemoryFreeClear(pBuf->assistBuf);
  taosMemoryFreeClear(pBuf);
}

SPageInfo* getLastPageInfo(SArray* pList) {
  size_t     size = taosArrayGetSize(pList);
  SPageInfo* pPgInfo = taosArrayGetP(pList, size - 1);
  return pPgInfo;
}

int32_t getPageId(const SPageInfo* pPgInfo) { return pPgInfo->pageId; }

int32_t getBufPageSize(const SDiskbasedBuf* pBuf) { return pBuf->pageSize; }

int32_t getNumOfInMemBufPages(const SDiskbasedBuf* pBuf) { return pBuf->inMemPages; }

bool isAllDataInMemBuf(const SDiskbasedBuf* pBuf) { return pBuf->fileSize == 0; }

void setBufPageDirty(void* pPage, bool dirty) {
  SPageInfo* ppi = getPageInfoFromPayload(pPage);
  ppi->dirty = dirty;
}

int32_t setBufPageCompressOnDisk(SDiskbasedBuf* pBuf, bool comp) {
  pBuf->comp = comp;
  if (comp && (pBuf->assistBuf == NULL)) {
    pBuf->assistBuf = taosMemoryMalloc(pBuf->pageSize + 2);  // EXTRA BYTES
    if (pBuf->assistBuf) {
      return terrno;
    }
  }
  return TSDB_CODE_SUCCESS;
}

int32_t dBufSetBufPageRecycled(SDiskbasedBuf* pBuf, void* pPage) {
  SPageInfo* ppi = getPageInfoFromPayload(pPage);

  int32_t code = tdListAppend(pBuf->freePgList, &ppi);
  if (TSDB_CODE_SUCCESS != code) {
    return code;
  }

  ppi->used = false;
  ppi->dirty = false;

  // add this pageinfo into the free page info list
  SListNode* pNode = tdListPopNode(pBuf->lruList, ppi->pn);
  taosMemoryFreeClear(ppi->pData);
  taosMemoryFreeClear(pNode);
  ppi->pn = NULL;
  return TSDB_CODE_SUCCESS;
}

void dBufSetPrintInfo(SDiskbasedBuf* pBuf) { pBuf->printStatis = true; }

SDiskbasedBufStatis getDBufStatis(const SDiskbasedBuf* pBuf) { return pBuf->statis; }

void dBufPrintStatis(const SDiskbasedBuf* pBuf) {
  if (!pBuf->printStatis) {
    return;
  }

  const SDiskbasedBufStatis* ps = &pBuf->statis;

#if 0
  printf(
      "Paged buffer closed, total:%.2f Kb (%d Pages), inmem size:%.2f Kb (%d Pages), file size:%.2f Kb, page size:%.2f "
      "Kb, %s\n",
      pBuf->totalBufSize / 1024.0, pBuf->numOfPages, listNEles(pBuf->lruList) * pBuf->pageSize / 1024.0,
      listNEles(pBuf->lruList), pBuf->fileSize / 1024.0, pBuf->pageSize / 1024.0f, pBuf->id);
#endif

  if (ps->loadPages > 0) {
    (void)printf(
        "Get/Release pages:%d/%d, flushToDisk:%.2f Kb (%d Pages), loadFromDisk:%.2f Kb (%d Pages), avgPageSize:%.2f "
        "Kb\n",
        ps->getPages, ps->releasePages, ps->flushBytes / 1024.0f, ps->flushPages, ps->loadBytes / 1024.0f,
        ps->loadPages, ps->loadBytes / (1024.0 * ps->loadPages));
  } else {
    // printf("no page loaded\n");
  }
}

void clearDiskbasedBuf(SDiskbasedBuf* pBuf) {
  if (pBuf == NULL) {
    return;
  }

  int64_t realSizeBefore = -1;
  if (pBuf->pFile != NULL && taosFStatFile(pBuf->pFile, &realSizeBefore, NULL) != TSDB_CODE_SUCCESS) {
    realSizeBefore = -1;
  }

  const SDiskbasedBufStatis* ps = &pBuf->statis;
  uDebug(
      "clear paged buffer begin, pages:%d, inMemPages:%d, fileSize:%" PRIu64 ", nextPos:%" PRIu64
      ", realSize:%" PRId64 ", get/release:%d/%d, flush/load:%d/%d, %s",
      pBuf->numOfPages, listNEles(pBuf->lruList), pBuf->fileSize, pBuf->nextPos, realSizeBefore, ps->getPages,
      ps->releasePages, ps->flushPages, ps->loadPages, pBuf->id);

  size_t n = taosArrayGetSize(pBuf->pIdList);
  for (int32_t i = 0; i < n; ++i) {
    SPageInfo* pi = taosArrayGetP(pBuf->pIdList, i);
    taosMemoryFreeClear(pi->pData);
    taosMemoryFreeClear(pi);
  }

  taosArrayClear(pBuf->pIdList);

  tdListEmpty(pBuf->lruList);
  tdListEmpty(pBuf->freePgList);

  taosArrayClear(pBuf->emptyDummyIdList);
  taosArrayClear(pBuf->pFree);

  tSimpleHashClear(pBuf->all);

  pBuf->numOfPages = 0;  // all pages are in buffer in the first place
  pBuf->totalBufSize = 0;
  pBuf->allocateId = -1;
  pBuf->fileSize = 0;
  pBuf->nextPos = 0;

  if (pBuf->pFile != NULL) {
    int32_t code = taosFtruncateFile(pBuf->pFile, 0);
    if (code != TSDB_CODE_SUCCESS) {
      uWarn("failed to truncate paged buffer file, path:%s, code:%s, %s", pBuf->path, tstrerror(code), pBuf->id);
    }
  }

  int64_t realSizeAfter = -1;
  if (pBuf->pFile != NULL && taosFStatFile(pBuf->pFile, &realSizeAfter, NULL) != TSDB_CODE_SUCCESS) {
    realSizeAfter = -1;
  }

  uDebug("clear paged buffer end, pages:%d, inMemPages:%d, fileSize:%" PRIu64 ", nextPos:%" PRIu64
         ", realSize:%" PRId64 ", %s",
         pBuf->numOfPages, listNEles(pBuf->lruList), pBuf->fileSize, pBuf->nextPos, realSizeAfter, pBuf->id);
}

taosdata / TDengine / #4999

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous