tpagedbuf.c 19.3 KB
Newer Older
S
Shengliang Guan 已提交
1
#define _DEFAULT_SOURCE
H
Haojun Liao 已提交
2
#include "tpagedbuf.h"
3 4
#include "taoserror.h"
#include "tcompression.h"
H
Haojun Liao 已提交
5
#include "thash.h"
S
Shengliang Guan 已提交
6
#include "tlog.h"
7

S
Shengliang Guan 已提交
8
#define GET_DATA_PAYLOAD(_p)          ((char*)(_p)->pData + POINTER_BYTES)
9 10
#define NO_IN_MEM_AVAILABLE_PAGES(_b) (listNEles((_b)->lruList) >= (_b)->inMemPages)

11
typedef struct SPageDiskInfo {
S
Shengliang Guan 已提交
12 13
  int64_t offset;
  int32_t length;
H
Haojun Liao 已提交
14
} SPageDiskInfo, SFreeListItem;
15

H
Haojun Liao 已提交
16
struct SPageInfo {
17
  SListNode* pn;  // point to list node struct
S
Shengliang Guan 已提交
18 19 20
  void*      pData;
  int64_t    offset;
  int32_t    pageId;
21
  int32_t    length : 29;
S
Shengliang Guan 已提交
22 23
  bool       used : 1;   // set current page is in used
  bool       dirty : 1;  // set current buffer page is dirty or not
H
Haojun Liao 已提交
24
};
25

H
Haojun Liao 已提交
26
struct SDiskbasedBuf {
27 28
  int32_t   numOfPages;
  int64_t   totalBufSize;
S
Shengliang Guan 已提交
29
  uint64_t  fileSize;  // disk file size
30
  TdFilePtr pFile;
S
Shengliang Guan 已提交
31 32 33 34
  int32_t   allocateId;  // allocated page id
  char*     path;        // file path
  int32_t   pageSize;    // current used page size
  int32_t   inMemPages;  // numOfPages that are allocated in memory
35
  SList*    freePgList;  // free page list
36
  SArray*   pIdList;     // page id list
37 38
  SHashObj* all;
  SList*    lruList;
S
Shengliang Guan 已提交
39 40 41 42 43 44
  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

H
Hongze Cheng 已提交
45 46
  char*               id;           // for debug purpose
  bool                printStatis;  // Print statistics info when closing this buffer.
H
Haojun Liao 已提交
47
  SDiskbasedBufStatis statis;
H
Haojun Liao 已提交
48
};
49

H
Haojun Liao 已提交
50
static int32_t createDiskFile(SDiskbasedBuf* pBuf) {
H
Hongze Cheng 已提交
51 52
  pBuf->pFile =
      taosOpenFile(pBuf->path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_TRUNC | TD_FILE_AUTO_DEL);
53
  if (pBuf->pFile == NULL) {
54 55 56 57 58 59
    return TAOS_SYSTEM_ERROR(errno);
  }

  return TSDB_CODE_SUCCESS;
}

S
Shengliang Guan 已提交
60
static char* doCompressData(void* data, int32_t srcSize, int32_t* dst, SDiskbasedBuf* pBuf) {  // do nothing
H
Haojun Liao 已提交
61
  if (!pBuf->comp) {
62 63 64 65
    *dst = srcSize;
    return data;
  }

H
Haojun Liao 已提交
66
  *dst = tsCompressString(data, srcSize, 1, pBuf->assistBuf, srcSize, ONE_STAGE_COMP, NULL, 0);
67

H
Haojun Liao 已提交
68
  memcpy(data, pBuf->assistBuf, *dst);
69 70 71
  return data;
}

S
Shengliang Guan 已提交
72
static char* doDecompressData(void* data, int32_t srcSize, int32_t* dst, SDiskbasedBuf* pBuf) {  // do nothing
H
Haojun Liao 已提交
73
  if (!pBuf->comp) {
74 75 76 77
    *dst = srcSize;
    return data;
  }

H
Haojun Liao 已提交
78
  *dst = tsDecompressString(data, srcSize, 1, pBuf->assistBuf, pBuf->pageSize, ONE_STAGE_COMP, NULL, 0);
79
  if (*dst > 0) {
H
Haojun Liao 已提交
80
    memcpy(data, pBuf->assistBuf, *dst);
81 82 83 84
  }
  return data;
}

H
Haojun Liao 已提交
85 86 87
static uint64_t allocatePositionInFile(SDiskbasedBuf* pBuf, size_t size) {
  if (pBuf->pFree == NULL) {
    return pBuf->nextPos;
88 89 90
  } else {
    int32_t offset = -1;

H
Haojun Liao 已提交
91
    size_t num = taosArrayGetSize(pBuf->pFree);
S
Shengliang Guan 已提交
92
    for (int32_t i = 0; i < num; ++i) {
H
Haojun Liao 已提交
93
      SFreeListItem* pi = taosArrayGet(pBuf->pFree, i);
H
Haojun Liao 已提交
94
      if (pi->length >= size) {
95 96
        offset = pi->offset;
        pi->offset += (int32_t)size;
H
Haojun Liao 已提交
97
        pi->length -= (int32_t)size;
98 99 100 101 102 103

        return offset;
      }
    }

    // no available recycle space, allocate new area in file
H
Haojun Liao 已提交
104
    return pBuf->nextPos;
105 106 107
  }
}

108
static void setPageNotInBuf(SPageInfo* pPageInfo) { pPageInfo->pData = NULL; }
H
Haojun Liao 已提交
109

110
static FORCE_INLINE size_t getAllocPageSize(int32_t pageSize) { return pageSize + POINTER_BYTES + sizeof(SFilePage); }
H
Haojun Liao 已提交
111

H
Haojun Liao 已提交
112 113 114 115 116 117 118 119
/**
 *   +--------------------------+-------------------+--------------+
 *   | PTR to SPageInfo (8bytes)| Payload (PageSize)| 2 Extra Bytes|
 *   +--------------------------+-------------------+--------------+
 * @param pBuf
 * @param pg
 * @return
 */
H
Haojun Liao 已提交
120
static char* doFlushPageToDisk(SDiskbasedBuf* pBuf, SPageInfo* pg) {
121 122
  assert(!pg->used && pg->pData != NULL);

H
Haojun Liao 已提交
123
  int32_t size = pBuf->pageSize;
H
Haojun Liao 已提交
124 125
  char*   t = NULL;
  if (pg->offset == -1 || pg->dirty) {
H
Haojun Liao 已提交
126 127 128
    void* payload = GET_DATA_PAYLOAD(pg);
    t = doCompressData(payload, pBuf->pageSize, &size, pBuf);
    assert(size >= 0);
H
Haojun Liao 已提交
129
  }
130 131

  // this page is flushed to disk for the first time
H
Haojun Liao 已提交
132 133 134
  if (pg->dirty) {
    if (pg->offset == -1) {
      assert(pg->dirty == true);
H
Haojun Liao 已提交
135

H
Haojun Liao 已提交
136 137
      pg->offset = allocatePositionInFile(pBuf, size);
      pBuf->nextPos += size;
138

139
      int32_t ret = taosLSeekFile(pBuf->pFile, pg->offset, SEEK_SET);
140
      if (ret == -1) {
141 142 143
        terrno = TAOS_SYSTEM_ERROR(errno);
        return NULL;
      }
144

145 146 147 148 149
      ret = (int32_t)taosWriteFile(pBuf->pFile, t, size);
      if (ret != size) {
        terrno = TAOS_SYSTEM_ERROR(errno);
        return NULL;
      }
150

H
Haojun Liao 已提交
151 152 153
      if (pBuf->fileSize < pg->offset + size) {
        pBuf->fileSize = pg->offset + size;
      }
H
Haojun Liao 已提交
154

H
Haojun Liao 已提交
155 156 157 158 159 160 161 162 163 164 165 166 167
      pBuf->statis.flushBytes += size;
      pBuf->statis.flushPages += 1;
    } 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 = pg->offset};
        taosArrayPush(pBuf->pFree, &dinfo);

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

169 170
      // 3. write to disk.
      int32_t ret = taosLSeekFile(pBuf->pFile, pg->offset, SEEK_SET);
171
      if (ret == -1) {
172 173 174
        terrno = TAOS_SYSTEM_ERROR(errno);
        return NULL;
      }
175

176 177 178 179 180
      ret = (int32_t)taosWriteFile(pBuf->pFile, t, size);
      if (ret != size) {
        terrno = TAOS_SYSTEM_ERROR(errno);
        return NULL;
      }
181

H
Haojun Liao 已提交
182 183 184
      if (pBuf->fileSize < pg->offset + size) {
        pBuf->fileSize = pg->offset + size;
      }
185

H
Haojun Liao 已提交
186 187
      pBuf->statis.flushBytes += size;
      pBuf->statis.flushPages += 1;
188
    }
189
  } else {  // NOTE: the size may be -1, the this recycle page has not been flushed to disk yet.
H
Haojun Liao 已提交
190
    size = pg->length;
191 192
  }

H
Haojun Liao 已提交
193
  ASSERT(size > 0 || (pg->offset == -1 && pg->length == -1));
194

H
Haojun Liao 已提交
195
  char* pDataBuf = pg->pData;
H
Haojun Liao 已提交
196
  memset(pDataBuf, 0, getAllocPageSize(pBuf->pageSize));
197
#ifdef BUF_PAGE_DEBUG
198
  uDebug("page_flush %p, pageId:%d, offset:%d", pDataBuf, pg->pageId, pg->offset);
199
#endif
H
Haojun Liao 已提交
200
  pg->length = size;  // on disk size
H
Haojun Liao 已提交
201
  return pDataBuf;
202 203
}

H
Haojun Liao 已提交
204
static char* flushPageToDisk(SDiskbasedBuf* pBuf, SPageInfo* pg) {
205
  int32_t ret = TSDB_CODE_SUCCESS;
S
Shengliang Guan 已提交
206
  assert(((int64_t)pBuf->numOfPages * pBuf->pageSize) == pBuf->totalBufSize && pBuf->numOfPages >= pBuf->inMemPages);
207

208
  if (pBuf->pFile == NULL) {
H
Haojun Liao 已提交
209
    if ((ret = createDiskFile(pBuf)) != TSDB_CODE_SUCCESS) {
210 211 212 213 214
      terrno = ret;
      return NULL;
    }
  }

H
Haojun Liao 已提交
215 216 217 218 219
  char* p = doFlushPageToDisk(pBuf, pg);
  setPageNotInBuf(pg);
  pg->dirty = false;

  return p;
220 221 222
}

// load file block data in disk
H
Haojun Liao 已提交
223
static int32_t loadPageFromDisk(SDiskbasedBuf* pBuf, SPageInfo* pg) {
224
  int32_t ret = taosLSeekFile(pBuf->pFile, pg->offset, SEEK_SET);
225
  if (ret == -1) {
H
Haojun Liao 已提交
226 227 228 229
    ret = TAOS_SYSTEM_ERROR(errno);
    return ret;
  }

230
  void* pPage = (void*)GET_DATA_PAYLOAD(pg);
H
Haojun Liao 已提交
231
  ret = (int32_t)taosReadFile(pBuf->pFile, pPage, pg->length);
232
  if (ret != pg->length) {
H
Haojun Liao 已提交
233 234
    ret = TAOS_SYSTEM_ERROR(errno);
    return ret;
235 236
  }

H
Haojun Liao 已提交
237 238
  pBuf->statis.loadBytes += pg->length;
  pBuf->statis.loadPages += 1;
239 240

  int32_t fullSize = 0;
H
Haojun Liao 已提交
241
  doDecompressData(pPage, pg->length, &fullSize, pBuf);
H
Haojun Liao 已提交
242
  return 0;
243 244
}

245
static SPageInfo* registerPage(SDiskbasedBuf* pBuf, int32_t pageId) {
H
Haojun Liao 已提交
246
  pBuf->numOfPages += 1;
247

wafwerar's avatar
wafwerar 已提交
248
  SPageInfo* ppi = taosMemoryMalloc(sizeof(SPageInfo));
249 250

  ppi->pageId = pageId;
H
Hongze Cheng 已提交
251
  ppi->pData = NULL;
252 253
  ppi->offset = -1;
  ppi->length = -1;
H
Hongze Cheng 已提交
254 255 256
  ppi->used = true;
  ppi->pn = NULL;
  ppi->dirty = false;
257

258
  return *(SPageInfo**)taosArrayPush(pBuf->pIdList, &ppi);
259 260
}

H
Haojun Liao 已提交
261
static SListNode* getEldestUnrefedPage(SDiskbasedBuf* pBuf) {
262
  SListIter iter = {0};
H
Haojun Liao 已提交
263
  tdListInitIter(pBuf->lruList, &iter, TD_LIST_BACKWARD);
264 265

  SListNode* pn = NULL;
S
Shengliang Guan 已提交
266 267
  while ((pn = tdListNext(&iter)) != NULL) {
    SPageInfo* pageInfo = *(SPageInfo**)pn->data;
268 269 270
    assert(pageInfo->pageId >= 0 && pageInfo->pn == pn);

    if (!pageInfo->used) {
271
      //      printf("%d is chosen\n", pageInfo->pageId);
272
      break;
H
Haojun Liao 已提交
273
    } else {
274
      //      printf("page %d is used, dirty:%d\n", pageInfo->pageId, pageInfo->dirty);
275 276 277 278 279 280
    }
  }

  return pn;
}

H
Haojun Liao 已提交
281
static char* evacOneDataPage(SDiskbasedBuf* pBuf) {
S
Shengliang Guan 已提交
282
  char*      bufPage = NULL;
H
Haojun Liao 已提交
283
  SListNode* pn = getEldestUnrefedPage(pBuf);
284
  terrno = 0;
285 286 287

  // all pages are referenced by user, try to allocate new space
  if (pn == NULL) {
H
Haojun Liao 已提交
288
    int32_t prev = pBuf->inMemPages;
289 290

    // increase by 50% of previous mem pages
H
Haojun Liao 已提交
291
    pBuf->inMemPages = (int32_t)(pBuf->inMemPages * 1.5f);
292

S
Shengliang Guan 已提交
293 294
    //    qWarn("%p in memory buf page not sufficient, expand from %d to %d, page size:%d", pBuf, prev,
    //          pBuf->inMemPages, pBuf->pageSize);
295
  } else {
H
Haojun Liao 已提交
296
    tdListPopNode(pBuf->lruList, pn);
297

S
Shengliang Guan 已提交
298
    SPageInfo* d = *(SPageInfo**)pn->data;
299 300 301
    assert(d->pn == pn);

    d->pn = NULL;
wafwerar's avatar
wafwerar 已提交
302
    taosMemoryFreeClear(pn);
303

H
Haojun Liao 已提交
304
    bufPage = flushPageToDisk(pBuf, d);
305 306 307 308 309
  }

  return bufPage;
}

S
Shengliang Guan 已提交
310
static void lruListPushFront(SList* pList, SPageInfo* pi) {
311 312 313 314 315
  tdListPrepend(pList, &pi);
  SListNode* front = tdListGetHead(pList);
  pi->pn = front;
}

S
Shengliang Guan 已提交
316
static void lruListMoveToFront(SList* pList, SPageInfo* pi) {
317 318 319 320
  tdListPopNode(pList, pi->pn);
  tdListPrependNode(pList, pi->pn);
}

H
Haojun Liao 已提交
321
static SPageInfo* getPageInfoFromPayload(void* page) {
H
Hongze Cheng 已提交
322
  char* p = (char*)page - POINTER_BYTES;
H
Haojun Liao 已提交
323

324
  SPageInfo* ppi = ((SPageInfo**)p)[0];
H
Haojun Liao 已提交
325
  return ppi;
326 327
}

H
Haojun Liao 已提交
328
int32_t createDiskbasedBuf(SDiskbasedBuf** pBuf, int32_t pagesize, int32_t inMemBufSize, const char* id,
329
                           const char* dir) {
wafwerar's avatar
wafwerar 已提交
330
  *pBuf = taosMemoryCalloc(1, sizeof(SDiskbasedBuf));
H
Haojun Liao 已提交
331

H
Haojun Liao 已提交
332 333
  SDiskbasedBuf* pPBuf = *pBuf;
  if (pPBuf == NULL) {
H
Haojun Liao 已提交
334 335 336
    return TSDB_CODE_OUT_OF_MEMORY;
  }

337
  pPBuf->pageSize = pagesize;
H
Hongze Cheng 已提交
338
  pPBuf->numOfPages = 0;  // all pages are in buffer in the first place
H
Haojun Liao 已提交
339
  pPBuf->totalBufSize = 0;
340 341
  pPBuf->inMemPages = inMemBufSize / pagesize;  // maximum allowed pages, it is a soft limit.
  pPBuf->allocateId = -1;
H
Hongze Cheng 已提交
342 343
  pPBuf->pFile = NULL;
  pPBuf->id = strdup(id);
344 345 346
  pPBuf->fileSize = 0;
  pPBuf->pFree = taosArrayInit(4, sizeof(SFreeListItem));
  pPBuf->freePgList = tdListNew(POINTER_BYTES);
H
Haojun Liao 已提交
347 348 349 350

  // at least more than 2 pages must be in memory
  assert(inMemBufSize >= pagesize * 2);

H
Haojun Liao 已提交
351
  pPBuf->lruList = tdListNew(POINTER_BYTES);
H
Haojun Liao 已提交
352 353 354

  // init id hash table
  _hash_fn_t fn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT);
355 356
  pPBuf->pIdList = taosArrayInit(4, POINTER_BYTES);

wafwerar's avatar
wafwerar 已提交
357
  pPBuf->assistBuf = taosMemoryMalloc(pPBuf->pageSize + 2);  // EXTRA BYTES
H
Haojun Liao 已提交
358
  pPBuf->all = taosHashInit(10, fn, true, false);
H
Haojun Liao 已提交
359 360 361

  char path[PATH_MAX] = {0};
  taosGetTmpfilePath(dir, "paged-buf", path);
H
Haojun Liao 已提交
362
  pPBuf->path = strdup(path);
H
Haojun Liao 已提交
363

H
Haojun Liao 已提交
364
  pPBuf->emptyDummyIdList = taosArrayInit(1, sizeof(int32_t));
H
Haojun Liao 已提交
365

366 367 368
  //  qDebug("QInfo:0x%"PRIx64" create resBuf for output, page size:%d, inmem buf pages:%d, file:%s", qId,
  //  pPBuf->pageSize,
  //         pPBuf->inMemPages, pPBuf->path);
H
Haojun Liao 已提交
369 370

  return TSDB_CODE_SUCCESS;
371 372
}

373
void* getNewBufPage(SDiskbasedBuf* pBuf, int32_t* pageId) {
H
Haojun Liao 已提交
374
  pBuf->statis.getPages += 1;
375 376

  char* availablePage = NULL;
H
Haojun Liao 已提交
377 378
  if (NO_IN_MEM_AVAILABLE_PAGES(pBuf)) {
    availablePage = evacOneDataPage(pBuf);
H
Haojun Liao 已提交
379 380
  }

H
Haojun Liao 已提交
381 382 383
  SPageInfo* pi = NULL;
  if (listNEles(pBuf->freePgList) != 0) {
    SListNode* pItem = tdListPopHead(pBuf->freePgList);
384
    pi = *(SPageInfo**)pItem->data;
H
Haojun Liao 已提交
385
    pi->used = true;
386
    *pageId = pi->pageId;
wafwerar's avatar
wafwerar 已提交
387
    taosMemoryFreeClear(pItem);
388
  } else {  // create a new pageinfo
H
Haojun Liao 已提交
389 390 391 392
    // register new id in this group
    *pageId = (++pBuf->allocateId);

    // register page id info
393
    pi = registerPage(pBuf, *pageId);
H
Haojun Liao 已提交
394 395 396 397 398

    // add to hash map
    taosHashPut(pBuf->all, pageId, sizeof(int32_t), &pi, POINTER_BYTES);
    pBuf->totalBufSize += pBuf->pageSize;
  }
399 400

  // add to LRU list
H
Haojun Liao 已提交
401 402
  assert(listNEles(pBuf->lruList) < pBuf->inMemPages && pBuf->inMemPages > 0);
  lruListPushFront(pBuf->lruList, pi);
403 404 405

  // allocate buf
  if (availablePage == NULL) {
H
Hongze Cheng 已提交
406 407
    pi->pData =
        taosMemoryCalloc(1, getAllocPageSize(pBuf->pageSize));  // add extract bytes in case of zipped buffer increased.
408 409 410 411 412
  } else {
    pi->pData = availablePage;
  }

  ((void**)pi->pData)[0] = pi;
413
#ifdef BUF_PAGE_DEBUG
H
Hongze Cheng 已提交
414
  uDebug("page_getNewBufPage , pi->pData:%p, pageId:%d, offset:%" PRId64, pi->pData, pi->pageId, pi->offset);
415
#endif
S
Shengliang Guan 已提交
416
  return (void*)(GET_DATA_PAYLOAD(pi));
417 418
}

H
Haojun Liao 已提交
419
void* getBufPage(SDiskbasedBuf* pBuf, int32_t id) {
H
Haojun Liao 已提交
420 421
  assert(pBuf != NULL && id >= 0);
  pBuf->statis.getPages += 1;
422

H
Haojun Liao 已提交
423
  SPageInfo** pi = taosHashGet(pBuf->all, &id, sizeof(int32_t));
424 425
  assert(pi != NULL && *pi != NULL);

S
Shengliang Guan 已提交
426
  if ((*pi)->pData != NULL) {  // it is in memory
427
    // no need to update the LRU list if only one page exists
H
Haojun Liao 已提交
428
    if (pBuf->numOfPages == 1) {
429
      (*pi)->used = true;
S
Shengliang Guan 已提交
430
      return (void*)(GET_DATA_PAYLOAD(*pi));
431 432
    }

S
Shengliang Guan 已提交
433
    SPageInfo** pInfo = (SPageInfo**)((*pi)->pn->data);
434 435
    assert(*pInfo == *pi);

H
Haojun Liao 已提交
436
    lruListMoveToFront(pBuf->lruList, (*pi));
437
    (*pi)->used = true;
438
#ifdef BUF_PAGE_DEBUG
H
Hongze Cheng 已提交
439
    uDebug("page_getBufPage1 pageId:%d, offset:%" PRId64, (*pi)->pageId, (*pi)->offset);
440
#endif
S
Shengliang Guan 已提交
441 442
    return (void*)(GET_DATA_PAYLOAD(*pi));
  } else {  // not in memory
H
Hongze Cheng 已提交
443 444
    assert((*pi)->pData == NULL && (*pi)->pn == NULL &&
           (((*pi)->length >= 0 && (*pi)->offset >= 0) || ((*pi)->length == -1 && (*pi)->offset == -1)));
445 446

    char* availablePage = NULL;
H
Haojun Liao 已提交
447 448
    if (NO_IN_MEM_AVAILABLE_PAGES(pBuf)) {
      availablePage = evacOneDataPage(pBuf);
449 450 451
      if (availablePage == NULL) {
        return NULL;
      }
452 453 454
    }

    if (availablePage == NULL) {
wafwerar's avatar
wafwerar 已提交
455
      (*pi)->pData = taosMemoryCalloc(1, getAllocPageSize(pBuf->pageSize));
456 457 458 459
    } else {
      (*pi)->pData = availablePage;
    }

H
Haojun Liao 已提交
460
    // set the ptr to the new SPageInfo
461 462
    ((void**)((*pi)->pData))[0] = (*pi);

H
Haojun Liao 已提交
463
    lruListPushFront(pBuf->lruList, *pi);
464 465
    (*pi)->used = true;

466 467 468 469 470 471
    // some data has been flushed to disk, and needs to be loaded into buffer again.
    if ((*pi)->length > 0 && (*pi)->offset >= 0) {
      int32_t code = loadPageFromDisk(pBuf, *pi);
      if (code != 0) {
        return NULL;
      }
H
Haojun Liao 已提交
472
    }
473
#ifdef BUF_PAGE_DEBUG
H
Hongze Cheng 已提交
474
    uDebug("page_getBufPage2 pageId:%d, offset:%" PRId64, (*pi)->pageId, (*pi)->offset);
475
#endif
S
Shengliang Guan 已提交
476
    return (void*)(GET_DATA_PAYLOAD(*pi));
477 478 479
  }
}

H
Haojun Liao 已提交
480 481
void releaseBufPage(SDiskbasedBuf* pBuf, void* page) {
  assert(pBuf != NULL && page != NULL);
H
Haojun Liao 已提交
482
  SPageInfo* ppi = getPageInfoFromPayload(page);
H
Haojun Liao 已提交
483
  releaseBufPageInfo(pBuf, ppi);
484 485
}

H
Haojun Liao 已提交
486
void releaseBufPageInfo(SDiskbasedBuf* pBuf, SPageInfo* pi) {
487
#ifdef BUF_PAGE_DEBUG
H
Hongze Cheng 已提交
488
  uDebug("page_releaseBufPageInfo pageId:%d, used:%d, offset:%" PRId64, pi->pageId, pi->used, pi->offset);
489
#endif
490 491
  // assert(pi->pData != NULL && pi->used == true);
  assert(pi->pData != NULL);
492
  pi->used = false;
H
Haojun Liao 已提交
493
  pBuf->statis.releasePages += 1;
494 495
}

H
Haojun Liao 已提交
496
size_t getTotalBufSize(const SDiskbasedBuf* pBuf) { return (size_t)pBuf->totalBufSize; }
497

498
SArray* getDataBufPagesIdList(SDiskbasedBuf* pBuf) {
499 500
  ASSERT(pBuf != NULL);
  return pBuf->pIdList;
501 502
}

H
Haojun Liao 已提交
503
void destroyDiskbasedBuf(SDiskbasedBuf* pBuf) {
H
Haojun Liao 已提交
504
  if (pBuf == NULL) {
505 506 507
    return;
  }

H
Haojun Liao 已提交
508
  dBufPrintStatis(pBuf);
H
Haojun Liao 已提交
509

510
  bool needRemoveFile = false;
511
  if (pBuf->pFile != NULL) {
512
    needRemoveFile = true;
S
Shengliang Guan 已提交
513 514
    uDebug(
        "Paged buffer closed, total:%.2f Kb (%d Pages), inmem size:%.2f Kb (%d Pages), file size:%.2f Kb, page "
H
Haojun Liao 已提交
515
        "size:%.2f Kb, %s\n",
S
Shengliang Guan 已提交
516
        pBuf->totalBufSize / 1024.0, pBuf->numOfPages, listNEles(pBuf->lruList) * pBuf->pageSize / 1024.0,
H
Haojun Liao 已提交
517
        listNEles(pBuf->lruList), pBuf->fileSize / 1024.0, pBuf->pageSize / 1024.0f, pBuf->id);
518

S
Shengliang Guan 已提交
519
    taosCloseFile(&pBuf->pFile);
520
  } else {
H
Haojun Liao 已提交
521
    uDebug("Paged buffer closed, total:%.2f Kb, no file created, %s", pBuf->totalBufSize / 1024.0, pBuf->id);
522 523
  }

H
Haojun Liao 已提交
524 525
  // print the statistics information
  {
S
Shengliang Guan 已提交
526
    SDiskbasedBufStatis* ps = &pBuf->statis;
527
    if (ps->loadPages == 0) {
H
Hongze Cheng 已提交
528 529
      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);
530 531
    } else {
      uDebug(
H
Hongze Cheng 已提交
532 533
          "Get/Release pages:%d/%d, flushToDisk:%.2f Kb (%d Pages), loadFromDisk:%.2f Kb (%d Pages), avgPageSize:%.2f "
          "Kb",
534 535 536
          ps->getPages, ps->releasePages, ps->flushBytes / 1024.0f, ps->flushPages, ps->loadBytes / 1024.0f,
          ps->loadPages, ps->loadBytes / (1024.0 * ps->loadPages));
    }
H
Haojun Liao 已提交
537
  }
538

539 540 541 542 543 544
  if (needRemoveFile) {
    int32_t ret = taosRemoveFile(pBuf->path);
    if (ret != 0) {  // print the error and discard this error info
      int32_t code = TAOS_SYSTEM_ERROR(errno);
      uDebug("WARNING tPage remove file failed. path=%s, code:%s", pBuf->path, strerror(code));
    }
545
  }
546

wafwerar's avatar
wafwerar 已提交
547
  taosMemoryFreeClear(pBuf->path);
H
Haojun Liao 已提交
548

549 550 551 552 553
  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);
554 555
  }

556 557
  taosArrayDestroy(pBuf->pIdList);

H
Haojun Liao 已提交
558
  tdListFree(pBuf->lruList);
H
Haojun Liao 已提交
559 560
  tdListFree(pBuf->freePgList);

H
Haojun Liao 已提交
561
  taosArrayDestroy(pBuf->emptyDummyIdList);
H
Haojun Liao 已提交
562 563
  taosArrayDestroy(pBuf->pFree);

H
Haojun Liao 已提交
564
  taosHashCleanup(pBuf->all);
565

wafwerar's avatar
wafwerar 已提交
566 567 568
  taosMemoryFreeClear(pBuf->id);
  taosMemoryFreeClear(pBuf->assistBuf);
  taosMemoryFreeClear(pBuf);
569 570
}

571
SPageInfo* getLastPageInfo(SArray* pList) {
S
Shengliang Guan 已提交
572
  size_t     size = taosArrayGetSize(pList);
573 574 575 576
  SPageInfo* pPgInfo = taosArrayGetP(pList, size - 1);
  return pPgInfo;
}

577
int32_t getPageId(const SPageInfo* pPgInfo) {
578 579 580 581
  ASSERT(pPgInfo != NULL);
  return pPgInfo->pageId;
}

S
Shengliang Guan 已提交
582
int32_t getBufPageSize(const SDiskbasedBuf* pBuf) { return pBuf->pageSize; }
H
Haojun Liao 已提交
583

S
Shengliang Guan 已提交
584
int32_t getNumOfInMemBufPages(const SDiskbasedBuf* pBuf) { return pBuf->inMemPages; }
H
Haojun Liao 已提交
585

S
Shengliang Guan 已提交
586
bool isAllDataInMemBuf(const SDiskbasedBuf* pBuf) { return pBuf->fileSize == 0; }
H
Haojun Liao 已提交
587

H
Haojun Liao 已提交
588
void setBufPageDirty(void* pPage, bool dirty) {
H
Haojun Liao 已提交
589
  SPageInfo* ppi = getPageInfoFromPayload(pPage);
H
Haojun Liao 已提交
590
  ppi->dirty = dirty;
591 592
}

593
void setBufPageCompressOnDisk(SDiskbasedBuf* pBuf, bool comp) { pBuf->comp = comp; }
H
Haojun Liao 已提交
594

595
void dBufSetBufPageRecycled(SDiskbasedBuf* pBuf, void* pPage) {
H
Haojun Liao 已提交
596 597
  SPageInfo* ppi = getPageInfoFromPayload(pPage);

598
  ppi->used = false;
H
Haojun Liao 已提交
599 600
  ppi->dirty = false;

H
Haojun Liao 已提交
601 602
  // add this pageinfo into the free page info list
  SListNode* pNode = tdListPopNode(pBuf->lruList, ppi->pn);
wafwerar's avatar
wafwerar 已提交
603 604
  taosMemoryFreeClear(ppi->pData);
  taosMemoryFreeClear(pNode);
605
  ppi->pn = NULL;
H
Haojun Liao 已提交
606 607

  tdListAppend(pBuf->freePgList, &ppi);
H
Haojun Liao 已提交
608
}
H
Haojun Liao 已提交
609

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

S
Shengliang Guan 已提交
612
SDiskbasedBufStatis getDBufStatis(const SDiskbasedBuf* pBuf) { return pBuf->statis; }
H
Haojun Liao 已提交
613

H
Haojun Liao 已提交
614
void dBufPrintStatis(const SDiskbasedBuf* pBuf) {
H
Haojun Liao 已提交
615 616 617 618 619 620 621 622
  if (!pBuf->printStatis) {
    return;
  }

  const SDiskbasedBufStatis* ps = &pBuf->statis;

  printf(
      "Paged buffer closed, total:%.2f Kb (%d Pages), inmem size:%.2f Kb (%d Pages), file size:%.2f Kb, page size:%.2f "
H
Haojun Liao 已提交
623
      "Kb, %s\n",
H
Haojun Liao 已提交
624
      pBuf->totalBufSize / 1024.0, pBuf->numOfPages, listNEles(pBuf->lruList) * pBuf->pageSize / 1024.0,
H
Haojun Liao 已提交
625
      listNEles(pBuf->lruList), pBuf->fileSize / 1024.0, pBuf->pageSize / 1024.0f, pBuf->id);
H
Haojun Liao 已提交
626

627 628
  if (ps->loadPages > 0) {
    printf(
H
Hongze Cheng 已提交
629 630
        "Get/Release pages:%d/%d, flushToDisk:%.2f Kb (%d Pages), loadFromDisk:%.2f Kb (%d Pages), avgPageSize:%.2f "
        "Kb\n",
631 632 633 634 635
        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");
  }
636
}
5
54liuyao 已提交
637 638

void clearDiskbasedBuf(SDiskbasedBuf* pBuf) {
639 640 641 642 643
  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);
5
54liuyao 已提交
644 645
  }

646
  taosArrayClear(pBuf->pIdList);
647

5
54liuyao 已提交
648 649 650 651 652 653 654 655
  tdListEmpty(pBuf->lruList);
  tdListEmpty(pBuf->freePgList);

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

  taosHashClear(pBuf->all);

H
Hongze Cheng 已提交
656
  pBuf->numOfPages = 0;  // all pages are in buffer in the first place
5
54liuyao 已提交
657 658 659
  pBuf->totalBufSize = 0;
  pBuf->allocateId = -1;
  pBuf->fileSize = 0;
660
}