• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

taosdata / TDengine / #3535

23 Nov 2024 02:07AM UTC coverage: 60.85% (+0.03%) from 60.825%
#3535

push

travis-ci

web-flow
Merge pull request #28893 from taosdata/doc/internal

refact: rename taos lib name

120252 of 252737 branches covered (47.58%)

Branch coverage included in aggregate %.

201187 of 275508 relevant lines covered (73.02%)

15886166.19 hits per line

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

61.66
/source/util/src/tpagedbuf.c
1
#define _DEFAULT_SOURCE
2
#include "tpagedbuf.h"
3
#include "taoserror.h"
4
#include "tcompression.h"
5
#include "tlog.h"
6
#include "tsimplehash.h"
7

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

14
typedef struct SPageDiskInfo {
15
  int64_t offset;
16
  int32_t length;
17
} SPageDiskInfo, SFreeListItem;
18

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

29
struct SDiskbasedBuf {
30
  int32_t    numOfPages;
31
  int64_t    totalBufSize;
32
  uint64_t   fileSize;  // disk file size
33
  TdFilePtr  pFile;
34
  int32_t    allocateId;  // allocated page id
35
  char*      path;        // file path
36
  char*      prefix;      // file name prefix
37
  int32_t    pageSize;    // current used page size
38
  int32_t    inMemPages;  // numOfPages that are allocated in memory
39
  SList*     freePgList;  // free page list
40
  SArray*    pIdList;     // page id list
41
  SSHashObj* all;
42
  SList*     lruList;
43
  void*      emptyDummyIdList;  // dummy id list
44
  void*      assistBuf;         // assistant buffer for compress/decompress data
45
  SArray*    pFree;             // free area in file
46
  bool       comp;              // compressed before flushed to disk
47
  uint64_t   nextPos;           // next page flush position
48

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

54
static int32_t createDiskFile(SDiskbasedBuf* pBuf) {
9,490✔
55
  if (pBuf->path == NULL) {  // prepare the file name when needed it
9,490!
56
    char path[PATH_MAX] = {0};
9,490✔
57
    taosGetTmpfilePath(pBuf->prefix, "paged-buf", path);
9,490✔
58
    pBuf->path = taosStrdup(path);
9,490✔
59
    if (pBuf->path == NULL) {
9,490!
60
      return terrno;
×
61
    }
62
  }
63

64
  pBuf->pFile =
9,490✔
65
      taosOpenFile(pBuf->path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_TRUNC | TD_FILE_AUTO_DEL);
9,490✔
66
  if (pBuf->pFile == NULL) {
9,490!
67
    return terrno;
×
68
  }
69

70
  return TSDB_CODE_SUCCESS;
9,490✔
71
}
72

73
static char* doCompressData(void* data, int32_t srcSize, int32_t* dst, SDiskbasedBuf* pBuf) {  // do nothing
9,662,086✔
74
  if (!pBuf->comp) {
9,662,086!
75
    *dst = srcSize;
9,662,226✔
76
    return data;
9,662,226✔
77
  }
78

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

81
  memcpy(data, pBuf->assistBuf, *dst);
×
82
  return data;
×
83
}
84

85
static int32_t doDecompressData(void* data, int32_t srcSize, int32_t* dst, SDiskbasedBuf* pBuf) {  // do nothing
8,815,332✔
86
  int32_t code = 0;
8,815,332✔
87
  if (!pBuf->comp) {
8,815,332!
88
    *dst = srcSize;
8,815,344✔
89
    return code;
8,815,344✔
90
  }
91

92
  *dst = tsDecompressString(data, srcSize, 1, pBuf->assistBuf, pBuf->pageSize, ONE_STAGE_COMP, NULL, 0);
×
93
  if (*dst > 0) {
×
94
    memcpy(data, pBuf->assistBuf, *dst);
×
95
  } else if (*dst < 0) {
×
96
    return terrno;
×
97
  }
98
  return code;
×
99
  ;
100
}
101

102
static uint64_t allocateNewPositionInFile(SDiskbasedBuf* pBuf, size_t size) {
8,888,435✔
103
  if (pBuf->pFree == NULL) {
8,888,435!
104
    return pBuf->nextPos;
×
105
  } else {
106
    int32_t offset = -1;
8,888,435✔
107

108
    size_t num = taosArrayGetSize(pBuf->pFree);
8,888,435✔
109
    for (int32_t i = 0; i < num; ++i) {
8,888,311!
110
      SFreeListItem* pi = taosArrayGet(pBuf->pFree, i);
×
111
      if (pi->length >= size) {
×
112
        offset = pi->offset;
×
113
        pi->offset += (int32_t)size;
×
114
        pi->length -= (int32_t)size;
×
115

116
        return offset;
×
117
      }
118
    }
119

120
    // no available recycle space, allocate new area in file
121
    return pBuf->nextPos;
8,888,311✔
122
  }
123
}
124

125
/**
126
 *   +--------------------------+-------------------+--------------+
127
 *   | PTR to SPageInfo (8bytes)| Payload (PageSize)| 2 Extra Bytes|
128
 *   +--------------------------+-------------------+--------------+
129
 * @param pBuf
130
 * @param pg
131
 * @return
132
 */
133

134
static FORCE_INLINE size_t getAllocPageSize(int32_t pageSize) { return pageSize + POINTER_BYTES + sizeof(SFilePage); }
38,491,545✔
135

136
static int32_t doFlushBufPageImpl(SDiskbasedBuf* pBuf, int64_t offset, const char* pData, int32_t size) {
9,661,972✔
137
  int64_t ret = taosLSeekFile(pBuf->pFile, offset, SEEK_SET);
9,661,972✔
138
  if (ret < 0) {
9,663,729!
139
    return terrno;
×
140
  }
141

142
  ret = (int32_t)taosWriteFile(pBuf->pFile, pData, size);
9,663,729✔
143
  if (ret != size) {
9,663,695!
144
    return terrno;
×
145
  }
146

147
  // extend the file
148
  if (pBuf->fileSize < offset + size) {
9,663,729✔
149
    pBuf->fileSize = offset + size;
8,889,963✔
150
  }
151

152
  pBuf->statis.flushBytes += size;
9,663,729✔
153
  pBuf->statis.flushPages += 1;
9,663,729✔
154

155
  return TSDB_CODE_SUCCESS;
9,663,729✔
156
}
157

158
static char* doFlushBufPage(SDiskbasedBuf* pBuf, SPageInfo* pg) {
15,443,486✔
159
  if (pg->pData == NULL || pg->used) {
15,443,486!
160
    uError("invalid params in paged buffer process when flushing buf to disk, %s", pBuf->id);
×
161
    terrno = TSDB_CODE_INVALID_PARA;
×
162
    return NULL;
×
163
  }
164

165
  int32_t size = pBuf->pageSize;
15,443,843✔
166
  int64_t offset = pg->offset;
15,443,843✔
167

168
  char* t = NULL;
15,443,843✔
169
  if ((!HAS_DATA_IN_DISK(pg)) || pg->dirty) {
15,443,843✔
170
    void* payload = GET_PAYLOAD_DATA(pg);
9,660,914✔
171
    t = doCompressData(payload, pBuf->pageSize + sizeof(SFilePage), &size, pBuf);
9,660,914✔
172
    if (size < 0) {
9,662,223!
173
      uError("failed to compress data when flushing data to disk, %s", pBuf->id);
×
174
      terrno = TSDB_CODE_INVALID_PARA;
×
175
      return NULL;
×
176
    }
177
  }
178

179
  // this page is flushed to disk for the first time
180
  if (pg->dirty) {
15,445,152✔
181
    if (!HAS_DATA_IN_DISK(pg)) {
9,662,231✔
182
      offset = allocateNewPositionInFile(pBuf, size);
8,888,502✔
183
      pBuf->nextPos += size;
8,888,250✔
184

185
      int32_t code = doFlushBufPageImpl(pBuf, offset, t, size);
8,888,250✔
186
      if (code != TSDB_CODE_SUCCESS) {
8,889,544!
187
        return NULL;
×
188
      }
189
    } else {
190
      // length becomes greater, current space is not enough, allocate new place, otherwise, do nothing
191
      if (pg->length < size) {
773,729!
192
        // 1. add current space to free list
193
        SPageDiskInfo dinfo = {.length = pg->length, .offset = offset};
×
194
        if (NULL == taosArrayPush(pBuf->pFree, &dinfo)) {
×
195
          return NULL;
×
196
        }
197

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

203
      int32_t code = doFlushBufPageImpl(pBuf, offset, t, size);
773,729✔
204
      if (code != TSDB_CODE_SUCCESS) {
773,755!
205
        return NULL;
×
206
      }
207
    }
208
  } else {  // NOTE: the size may be -1, the this recycle page has not been flushed to disk yet.
209
    size = pg->length;
5,782,921✔
210
  }
211

212
  char* pDataBuf = pg->pData;
15,446,220✔
213
  memset(pDataBuf, 0, getAllocPageSize(pBuf->pageSize));
15,446,220✔
214

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

219
  pg->offset = offset;
15,446,220✔
220
  pg->length = size;  // on disk size
15,446,220✔
221
  return pDataBuf;
15,446,220✔
222
}
223

224
static char* flushBufPage(SDiskbasedBuf* pBuf, SPageInfo* pg) {
15,443,369✔
225
  int32_t ret = TSDB_CODE_SUCCESS;
15,443,369✔
226

227
  if (pBuf->pFile == NULL) {
15,443,369✔
228
    if ((ret = createDiskFile(pBuf)) != TSDB_CODE_SUCCESS) {
9,490!
229
      terrno = ret;
×
230
      return NULL;
×
231
    }
232
  }
233

234
  char* p = doFlushBufPage(pBuf, pg);
15,443,369✔
235
  CLEAR_BUF_PAGE_IN_MEM_FLAG(pg);
15,445,009✔
236

237
  pg->dirty = false;
15,445,009✔
238
  return p;
15,445,009✔
239
}
240

241
// load file block data in disk
242
static int32_t loadPageFromDisk(SDiskbasedBuf* pBuf, SPageInfo* pg) {
8,813,960✔
243
  if (pg->offset < 0 || pg->length <= 0) {
8,813,960!
244
    uError("failed to load buf page from disk, offset:%" PRId64 ", length:%d, %s", pg->offset, pg->length, pBuf->id);
×
245
    return TSDB_CODE_INVALID_PARA;
×
246
  }
247

248
  int64_t ret = taosLSeekFile(pBuf->pFile, pg->offset, SEEK_SET);
8,815,499✔
249
  if (ret < 0) {
8,815,375!
250
    ret = terrno;
×
251
    return ret;
×
252
  }
253

254
  void* pPage = (void*)GET_PAYLOAD_DATA(pg);
8,815,375✔
255
  ret = taosReadFile(pBuf->pFile, pPage, pg->length);
8,815,375✔
256
  if (ret != pg->length) {
8,815,398!
257
    ret = terrno;
×
258
    return ret;
×
259
  }
260

261
  pBuf->statis.loadBytes += pg->length;
8,815,398✔
262
  pBuf->statis.loadPages += 1;
8,815,398✔
263

264
  int32_t fullSize = 0;
8,815,398✔
265
  return doDecompressData(pPage, pg->length, &fullSize, pBuf);
8,815,398✔
266
}
267

268
static SPageInfo* registerNewPageInfo(SDiskbasedBuf* pBuf, int32_t pageId) {
29,691,074✔
269
  pBuf->numOfPages += 1;
29,691,074✔
270

271
  SPageInfo* ppi = taosMemoryMalloc(sizeof(SPageInfo));
29,691,074✔
272
  if (ppi == NULL) {
29,710,597!
273
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
274
    return NULL;
×
275
  }
276

277
  ppi->pageId = pageId;
29,710,597✔
278
  ppi->pData = NULL;
29,710,597✔
279
  ppi->offset = -1;
29,710,597✔
280
  ppi->length = -1;
29,710,597✔
281
  ppi->used = true;
29,710,597✔
282
  ppi->pn = NULL;
29,710,597✔
283
  ppi->dirty = false;
29,710,597✔
284

285
  SPageInfo** pRet = taosArrayPush(pBuf->pIdList, &ppi);
29,710,597✔
286
  if (NULL == pRet) {
29,706,659!
287
    taosMemoryFree(ppi);
×
288
    return NULL;
×
289
  }
290
  return *pRet;
29,706,659✔
291
}
292

293
static SListNode* getEldestUnrefedPage(SDiskbasedBuf* pBuf) {
15,444,999✔
294
  SListIter iter = {0};
15,444,999✔
295
  tdListInitIter(pBuf->lruList, &iter, TD_LIST_BACKWARD);
15,444,999✔
296

297
  SListNode* pn = NULL;
15,444,870✔
298
  while ((pn = tdListNext(&iter)) != NULL) {
17,923,447✔
299
    SPageInfo* pageInfo = *(SPageInfo**)pn->data;
17,923,613✔
300

301
    SPageInfo* p = *(SPageInfo**)(pageInfo->pData);
17,923,613✔
302

303
    if (!pageInfo->used) {
17,923,613✔
304
      break;
15,445,036✔
305
    }
306
  }
307

308
  return pn;
15,445,050✔
309
}
310

311
static char* evictBufPage(SDiskbasedBuf* pBuf) {
15,445,005✔
312
  SListNode* pn = getEldestUnrefedPage(pBuf);
15,445,005✔
313
  if (pn == NULL) {  // no available buffer pages now, return.
15,444,946!
314
    return NULL;
×
315
  }
316

317
  terrno = 0;
15,444,946✔
318
  pn = tdListPopNode(pBuf->lruList, pn);
15,444,763✔
319

320
  SPageInfo* d = *(SPageInfo**)pn->data;
15,442,275✔
321

322
  d->pn = NULL;
15,442,275✔
323
  taosMemoryFreeClear(pn);
15,442,275!
324

325
  return flushBufPage(pBuf, d);
15,442,859✔
326
}
327

328
static int32_t lruListPushFront(SList* pList, SPageInfo* pi) {
38,519,508✔
329
  int32_t code = tdListPrepend(pList, &pi);
38,519,508✔
330
  if (TSDB_CODE_SUCCESS != code) {
38,521,038!
331
    return code;
×
332
  }
333
  SListNode* front = tdListGetHead(pList);
38,521,038✔
334
  pi->pn = front;
38,516,865✔
335
  return TSDB_CODE_SUCCESS;
38,516,865✔
336
}
337

338
static void lruListMoveToFront(SList* pList, SPageInfo* pi) {
1,066,609,902✔
339
  pi->pn = tdListPopNode(pList, pi->pn);
1,066,609,902✔
340
  tdListPrependNode(pList, pi->pn);
1,065,824,743✔
341
}
1,066,224,631✔
342

343
static SPageInfo* getPageInfoFromPayload(void* page) {
1,525,034,921✔
344
  char* p = (char*)page - POINTER_BYTES;
1,525,034,921✔
345

346
  SPageInfo* ppi = ((SPageInfo**)p)[0];
1,525,034,921✔
347
  return ppi;
1,525,034,921✔
348
}
349

350
int32_t createDiskbasedBuf(SDiskbasedBuf** pBuf, int32_t pagesize, int32_t inMemBufSize, const char* id,
9,104,451✔
351
                           const char* dir) {
352
  *pBuf = NULL;
9,104,451✔
353
  SDiskbasedBuf* pPBuf = taosMemoryCalloc(1, sizeof(SDiskbasedBuf));
9,104,451✔
354
  if (pPBuf == NULL) {
9,106,693!
355
    goto _error;
×
356
  }
357

358
  pPBuf->pageSize = pagesize;
9,106,693✔
359
  pPBuf->numOfPages = 0;  // all pages are in buffer in the first place
9,106,693✔
360
  pPBuf->totalBufSize = 0;
9,106,693✔
361
  pPBuf->allocateId = -1;
9,106,693✔
362
  pPBuf->pFile = NULL;
9,106,693✔
363
  pPBuf->id = taosStrdup(id);
9,106,693✔
364
  if (id != NULL && pPBuf->id == NULL) {
9,109,477!
365
    goto _error;
×
366
  }
367
  pPBuf->fileSize = 0;
9,109,477✔
368
  pPBuf->pFree = taosArrayInit(4, sizeof(SFreeListItem));
9,109,477✔
369
  pPBuf->freePgList = tdListNew(POINTER_BYTES);
9,112,424✔
370
  if (pPBuf->pFree == NULL || pPBuf->freePgList == NULL) {
9,106,287!
371
    goto _error;
×
372
  }
373

374
  // at least more than 2 pages must be in memory
375
  if (inMemBufSize < pagesize * 2) {
9,106,287✔
376
    inMemBufSize = pagesize * 2;
4,322✔
377
  }
378

379
  pPBuf->inMemPages = inMemBufSize / pagesize;  // maximum allowed pages, it is a soft limit.
9,106,287✔
380
  pPBuf->lruList = tdListNew(POINTER_BYTES);
9,106,287✔
381
  if (pPBuf->lruList == NULL) {
9,109,727!
382
    goto _error;
×
383
  }
384

385
  // init id hash table
386
  _hash_fn_t fn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT);
9,109,727✔
387
  pPBuf->pIdList = taosArrayInit(4, POINTER_BYTES);
9,109,058✔
388
  if (pPBuf->pIdList == NULL) {
9,112,074!
389
    goto _error;
×
390
  }
391

392
  pPBuf->all = tSimpleHashInit(64, fn);
9,112,074✔
393
  if (pPBuf->all == NULL) {
9,105,679!
394
    goto _error;
×
395
  }
396

397
  pPBuf->prefix = (char*)dir;
9,105,679✔
398
  pPBuf->emptyDummyIdList = taosArrayInit(1, sizeof(int32_t));
9,105,679✔
399
  if (pPBuf->emptyDummyIdList == NULL) {
9,112,633!
400
    goto _error;
×
401
  }
402

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

406
  *pBuf = pPBuf;
9,112,633✔
407
  return TSDB_CODE_SUCCESS;
9,112,633✔
408

409
_error:
×
410
  destroyDiskbasedBuf(pPBuf);
×
411
  *pBuf = NULL;
×
412
  return TSDB_CODE_OUT_OF_MEMORY;
×
413
}
414

415
static char* doExtractPage(SDiskbasedBuf* pBuf) {
38,490,357✔
416
  char* availablePage = NULL;
38,490,357✔
417
  if (NO_IN_MEM_AVAILABLE_PAGES(pBuf)) {
38,490,357✔
418
    availablePage = evictBufPage(pBuf);
15,445,032✔
419
    if (availablePage == NULL) {
15,445,028!
420
      uWarn("no available buf pages, current:%d, max:%d, reason: %s, %s", listNEles(pBuf->lruList), pBuf->inMemPages,
×
421
            terrstr(), pBuf->id)
422
    }
423
  } else {
424
    availablePage =
425
        taosMemoryCalloc(1, getAllocPageSize(pBuf->pageSize));  // add extract bytes in case of zipped buffer increased.
46,090,650✔
426
    if (availablePage == NULL) {
23,058,416!
427
      terrno = TSDB_CODE_OUT_OF_MEMORY;
×
428
    }
429
  }
430

431
  return availablePage;
38,500,525✔
432
}
433

434
void* getNewBufPage(SDiskbasedBuf* pBuf, int32_t* pageId) {
29,676,842✔
435
  pBuf->statis.getPages += 1;
29,676,842✔
436

437
  char* availablePage = doExtractPage(pBuf);
29,676,842✔
438
  if (availablePage == NULL) {
29,689,517!
439
    return NULL;
×
440
  }
441

442
  SPageInfo* pi = NULL;
29,689,517✔
443
  int32_t    code = 0;
29,689,517✔
444
  if (listNEles(pBuf->freePgList) != 0) {
29,689,517!
445
    SListNode* pItem = tdListPopHead(pBuf->freePgList);
×
446
    pi = *(SPageInfo**)pItem->data;
×
447
    pi->used = true;
×
448
    *pageId = pi->pageId;
×
449
    taosMemoryFreeClear(pItem);
×
450
    code = lruListPushFront(pBuf->lruList, pi);
×
451
    if (TSDB_CODE_SUCCESS != code) {
×
452
      taosMemoryFree(pi);
459✔
453
      taosMemoryFree(availablePage);
×
454
      terrno = code;
×
455
      return NULL;
×
456
    }
457
  } else {  // create a new pageinfo
458
    // register new id in this group
459
    *pageId = (++pBuf->allocateId);
29,689,517✔
460

461
    // register page id info
462
    pi = registerNewPageInfo(pBuf, *pageId);
29,689,517✔
463
    if (pi == NULL) {
29,704,031!
464
      taosMemoryFree(availablePage);
×
465
      return NULL;
×
466
    }
467

468
    // add to hash map
469
    int32_t code = tSimpleHashPut(pBuf->all, pageId, sizeof(int32_t), &pi, POINTER_BYTES);
29,704,031✔
470

471
    if (TSDB_CODE_SUCCESS == code) {
29,709,483!
472
      // add to LRU list
473
      code = lruListPushFront(pBuf->lruList, pi);
29,709,524✔
474
    }
475
    if (TSDB_CODE_SUCCESS == code) {
29,704,396!
476
      pBuf->totalBufSize += pBuf->pageSize;
29,704,396✔
477
    } else {
478
      taosMemoryFree(availablePage);
×
479
      SPageInfo **pLast = taosArrayPop(pBuf->pIdList);
×
480
      int32_t ret = tSimpleHashRemove(pBuf->all, pageId, sizeof(int32_t));
×
481
      if (ret != TSDB_CODE_SUCCESS) {
×
482
        uError("%s failed to clear pageId %d from buf hash-set since %s", __func__, *pageId, tstrerror(ret));
×
483
      }
484
      taosMemoryFree(pi);
×
485
      terrno = code;
×
486
      return NULL;
×
487
    }
488
  }
489

490
  pi->pData = availablePage;
29,703,937✔
491

492
  ((void**)pi->pData)[0] = pi;
29,703,937✔
493
#ifdef BUF_PAGE_DEBUG
494
  uDebug("page_getNewBufPage , pi->pData:%p, pageId:%d, offset:%" PRId64, pi->pData, pi->pageId, pi->offset);
495
#endif
496

497
  return (void*)(GET_PAYLOAD_DATA(pi));
29,703,937✔
498
}
499

500
void* getBufPage(SDiskbasedBuf* pBuf, int32_t id) {
1,212,608,451✔
501
  if (id < 0) {
1,212,608,451!
502
    terrno = TSDB_CODE_INVALID_PARA;
×
503
    uError("invalid page id:%d, %s", id, pBuf->id);
×
504
    return NULL;
×
505
  }
506

507
  pBuf->statis.getPages += 1;
1,212,608,451✔
508

509
  SPageInfo** pi = tSimpleHashGet(pBuf->all, &id, sizeof(int32_t));
1,212,608,451✔
510
  if (pi == NULL || *pi == NULL) {
1,211,996,900!
511
    uError("failed to locate the buffer page:%d, %s", id, pBuf->id);
×
512
    terrno = TSDB_CODE_INVALID_PARA;
×
513
    return NULL;
×
514
  }
515

516
  if (BUF_PAGE_IN_MEM(*pi)) {  // it is in memory
1,212,630,283✔
517
    // no need to update the LRU list if only one page exists
518
    if (pBuf->numOfPages == 1) {
1,203,607,074✔
519
      (*pi)->used = true;
138,240,611✔
520
      return (void*)(GET_PAYLOAD_DATA(*pi));
138,240,611✔
521
    }
522

523
    SPageInfo** pInfo = (SPageInfo**)((*pi)->pn->data);
1,065,366,463✔
524
    if (*pInfo != *pi) {
1,065,366,463!
525
      terrno = TSDB_CODE_APP_ERROR;
×
526
      uError("inconsistently data in paged buffer, pInfo:%p, pi:%p, %s", *pInfo, *pi, pBuf->id);
×
527
      return NULL;
×
528
    }
529

530
    lruListMoveToFront(pBuf->lruList, (*pi));
1,065,366,463✔
531
    (*pi)->used = true;
1,066,270,663✔
532

533
#ifdef BUF_PAGE_DEBUG
534
    uDebug("page_getBufPage1 pageId:%d, offset:%" PRId64, (*pi)->pageId, (*pi)->offset);
535
#endif
536
    return (void*)(GET_PAYLOAD_DATA(*pi));
1,066,270,663✔
537
  } else {  // not in memory
538

539
    (*pi)->pData = doExtractPage(pBuf);
9,023,209✔
540

541
    // failed to evict buffer page, return with error code.
542
    if ((*pi)->pData == NULL) {
8,813,688!
543
      return NULL;
×
544
    }
545

546
    // set the ptr to the new SPageInfo
547
    ((void**)((*pi)->pData))[0] = (*pi);
8,813,688✔
548

549
    int32_t code = lruListPushFront(pBuf->lruList, *pi);
8,813,688✔
550
    if (TSDB_CODE_SUCCESS != code) {
8,813,911!
551
      taosMemoryFree((*pi)->pData);
×
552
      (*pi)->pData = NULL;
×
553
      terrno = code;
×
554
      return NULL;
×
555
    }
556
    (*pi)->used = true;
8,813,911✔
557

558
    // some data has been flushed to disk, and needs to be loaded into buffer again.
559
    if (HAS_DATA_IN_DISK(*pi)) {
8,813,911!
560
      int32_t code = loadPageFromDisk(pBuf, *pi);
8,813,955✔
561
      if (code != 0) {
8,815,235!
562
        taosMemoryFree((*pi)->pData);
×
563
        (*pi)->pData = NULL;
×
564
        terrno = code;
×
565
        return NULL;
×
566
      }
567
    }
568
#ifdef BUF_PAGE_DEBUG
569
    uDebug("page_getBufPage2 pageId:%d, offset:%" PRId64, (*pi)->pageId, (*pi)->offset);
570
#endif
571
    return (void*)(GET_PAYLOAD_DATA(*pi));
8,815,271✔
572
  }
573
}
574

575
void releaseBufPage(SDiskbasedBuf* pBuf, void* page) {
966,312,647✔
576
  if (page == NULL) {
966,312,647!
577
    return;
×
578
  }
579

580
  SPageInfo* ppi = getPageInfoFromPayload(page);
966,312,647✔
581
  releaseBufPageInfo(pBuf, ppi);
964,314,976✔
582
}
583

584
void releaseBufPageInfo(SDiskbasedBuf* pBuf, SPageInfo* pi) {
964,185,273✔
585
#ifdef BUF_PAGE_DEBUG
586
  uDebug("page_releaseBufPageInfo pageId:%d, used:%d, offset:%" PRId64, pi->pageId, pi->used, pi->offset);
587
#endif
588

589
  if (pi == NULL) {
964,185,273!
590
    return;
×
591
  }
592

593
  if (pi->pData == NULL) {
964,185,273!
594
    uError("pi->pData (page data) is null");
×
595
    return;
×
596
  }
597

598
  pi->used = false;
964,185,273✔
599
  pBuf->statis.releasePages += 1;
964,185,273✔
600
}
601

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

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

606
void destroyDiskbasedBuf(SDiskbasedBuf* pBuf) {
11,736,579✔
607
  if (pBuf == NULL) {
11,736,579✔
608
    return;
2,624,334✔
609
  }
610

611
  dBufPrintStatis(pBuf);
9,112,245✔
612

613
  bool needRemoveFile = false;
9,112,332✔
614
  if (pBuf->pFile != NULL) {
9,112,332✔
615
    needRemoveFile = true;
9,489✔
616
    uDebug(
9,489✔
617
        "Paged buffer closed, total:%.2f Kb (%d Pages), inmem size:%.2f Kb (%d Pages), file size:%.2f Kb, page "
618
        "size:%.2f Kb, %s",
619
        pBuf->totalBufSize / 1024.0, pBuf->numOfPages, listNEles(pBuf->lruList) * pBuf->pageSize / 1024.0,
620
        listNEles(pBuf->lruList), pBuf->fileSize / 1024.0, pBuf->pageSize / 1024.0f, pBuf->id);
621

622
    int32_t code = taosCloseFile(&pBuf->pFile);
9,489✔
623
    if (TSDB_CODE_SUCCESS != code) {
9,490!
624
      uDebug("WARNING tPage failed to close file when destroy disk basebuf: %s", pBuf->path);
×
625
    }
626
  } else {
627
    uDebug("Paged buffer closed, total:%.2f Kb, no file created, %s", pBuf->totalBufSize / 1024.0, pBuf->id);
9,102,843✔
628
  }
629

630
  // print the statistics information
631
  {
632
    SDiskbasedBufStatis* ps = &pBuf->statis;
9,112,154✔
633
    if (ps->loadPages == 0) {
9,112,154✔
634
      uDebug("Get/Release pages:%d/%d, flushToDisk:%.2f Kb (%d Pages), loadFromDisk:%.2f Kb (%d Pages)", ps->getPages,
9,102,664✔
635
             ps->releasePages, ps->flushBytes / 1024.0f, ps->flushPages, ps->loadBytes / 1024.0f, ps->loadPages);
636
    } else {
637
      uDebug(
9,490✔
638
          "Get/Release pages:%d/%d, flushToDisk:%.2f Kb (%d Pages), loadFromDisk:%.2f Kb (%d Pages), avgPgSize:%.2f Kb",
639
          ps->getPages, ps->releasePages, ps->flushBytes / 1024.0f, ps->flushPages, ps->loadBytes / 1024.0f,
640
          ps->loadPages, ps->loadBytes / (1024.0 * ps->loadPages));
641
    }
642
  }
643

644
  if (needRemoveFile) {
9,112,154✔
645
    int32_t ret = taosRemoveFile(pBuf->path);
9,490✔
646
    if (ret != 0) {  // print the error and discard this error info
9,490!
647
      uDebug("WARNING tPage remove file failed. path=%s, code:%s", pBuf->path, strerror(errno));
9,490✔
648
    }
649
  }
650

651
  taosMemoryFreeClear(pBuf->path);
9,111,912✔
652

653
  size_t n = taosArrayGetSize(pBuf->pIdList);
9,111,912✔
654
  for (int32_t i = 0; i < n; ++i) {
37,519,532✔
655
    SPageInfo* pi = taosArrayGetP(pBuf->pIdList, i);
28,407,824✔
656
    taosMemoryFreeClear(pi->pData);
28,357,391✔
657
    taosMemoryFreeClear(pi);
28,422,318✔
658
  }
659

660
  taosArrayDestroy(pBuf->pIdList);
9,111,708✔
661

662
  pBuf->lruList = tdListFree(pBuf->lruList);
9,112,528✔
663
  pBuf->freePgList = tdListFree(pBuf->freePgList);
9,112,662✔
664

665
  taosArrayDestroy(pBuf->emptyDummyIdList);
9,112,813✔
666
  taosArrayDestroy(pBuf->pFree);
9,112,662✔
667

668
  tSimpleHashCleanup(pBuf->all);
9,112,740✔
669

670
  taosMemoryFreeClear(pBuf->id);
9,112,884!
671
  taosMemoryFreeClear(pBuf->assistBuf);
9,112,909!
672
  taosMemoryFreeClear(pBuf);
9,112,909✔
673
}
674

675
SPageInfo* getLastPageInfo(SArray* pList) {
×
676
  size_t     size = taosArrayGetSize(pList);
×
677
  SPageInfo* pPgInfo = taosArrayGetP(pList, size - 1);
×
678
  return pPgInfo;
×
679
}
680

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

683
int32_t getBufPageSize(const SDiskbasedBuf* pBuf) { return pBuf->pageSize; }
579,287,264✔
684

685
int32_t getNumOfInMemBufPages(const SDiskbasedBuf* pBuf) { return pBuf->inMemPages; }
2,236,894✔
686

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

689
void setBufPageDirty(void* pPage, bool dirty) {
573,870,335✔
690
  SPageInfo* ppi = getPageInfoFromPayload(pPage);
573,870,335✔
691
  ppi->dirty = dirty;
573,342,915✔
692
}
573,342,915✔
693

694
int32_t setBufPageCompressOnDisk(SDiskbasedBuf* pBuf, bool comp) {
×
695
  pBuf->comp = comp;
×
696
  if (comp && (pBuf->assistBuf == NULL)) {
×
697
    pBuf->assistBuf = taosMemoryMalloc(pBuf->pageSize + 2);  // EXTRA BYTES
×
698
    if (pBuf->assistBuf) {
×
699
      return terrno;
×
700
    }
701
  }
702
  return TSDB_CODE_SUCCESS;
×
703
}
704

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

708
  int32_t code = tdListAppend(pBuf->freePgList, &ppi);
×
709
  if (TSDB_CODE_SUCCESS != code) {
×
710
    return code;
×
711
  }
712

713
  ppi->used = false;
×
714
  ppi->dirty = false;
×
715

716
  // add this pageinfo into the free page info list
717
  SListNode* pNode = tdListPopNode(pBuf->lruList, ppi->pn);
×
718
  taosMemoryFreeClear(ppi->pData);
×
719
  taosMemoryFreeClear(pNode);
×
720
  ppi->pn = NULL;
×
721
  return TSDB_CODE_SUCCESS;
×
722
}
723

724
void dBufSetPrintInfo(SDiskbasedBuf* pBuf) { pBuf->printStatis = true; }
3,631,068✔
725

726
SDiskbasedBufStatis getDBufStatis(const SDiskbasedBuf* pBuf) { return pBuf->statis; }
2,369,909✔
727

728
void dBufPrintStatis(const SDiskbasedBuf* pBuf) {
9,112,238✔
729
  if (!pBuf->printStatis) {
9,112,238✔
730
    return;
5,480,384✔
731
  }
732

733
  const SDiskbasedBufStatis* ps = &pBuf->statis;
3,631,854✔
734

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

743
  if (ps->loadPages > 0) {
3,631,854✔
744
    (void)printf(
437✔
745
        "Get/Release pages:%d/%d, flushToDisk:%.2f Kb (%d Pages), loadFromDisk:%.2f Kb (%d Pages), avgPageSize:%.2f "
746
        "Kb\n",
747
        ps->getPages, ps->releasePages, ps->flushBytes / 1024.0f, ps->flushPages, ps->loadBytes / 1024.0f,
437✔
748
        ps->loadPages, ps->loadBytes / (1024.0 * ps->loadPages));
437✔
749
  } else {
750
    // printf("no page loaded\n");
751
  }
752
}
753

754
void clearDiskbasedBuf(SDiskbasedBuf* pBuf) {
29,015✔
755
  size_t n = taosArrayGetSize(pBuf->pIdList);
29,015✔
756
  for (int32_t i = 0; i < n; ++i) {
1,285,836✔
757
    SPageInfo* pi = taosArrayGetP(pBuf->pIdList, i);
1,256,827✔
758
    taosMemoryFreeClear(pi->pData);
1,256,749✔
759
    taosMemoryFreeClear(pi);
1,256,807✔
760
  }
761

762
  taosArrayClear(pBuf->pIdList);
29,009✔
763

764
  tdListEmpty(pBuf->lruList);
29,014✔
765
  tdListEmpty(pBuf->freePgList);
29,014✔
766

767
  taosArrayClear(pBuf->emptyDummyIdList);
29,015✔
768
  taosArrayClear(pBuf->pFree);
29,014✔
769

770
  tSimpleHashClear(pBuf->all);
29,013✔
771

772
  pBuf->numOfPages = 0;  // all pages are in buffer in the first place
29,015✔
773
  pBuf->totalBufSize = 0;
29,015✔
774
  pBuf->allocateId = -1;
29,015✔
775
  pBuf->fileSize = 0;
29,015✔
776
}
29,015✔
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc