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

taosdata / TDengine / #4829

30 Oct 2025 09:25AM UTC coverage: 49.734% (-11.3%) from 61.071%
#4829

push

travis-ci

web-flow
Merge pull request #33435 from taosdata/3.0

merge 3.0

123072 of 323930 branches covered (37.99%)

Branch coverage included in aggregate %.

7 of 25 new or added lines in 3 files covered. (28.0%)

35232 existing lines in 327 files now uncovered.

172062 of 269495 relevant lines covered (63.85%)

70709785.06 hits per line

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

55.5
/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) {
38,584✔
55
  if (pBuf->path == NULL) {  // prepare the file name when needed it
38,584!
56
    char path[PATH_MAX] = {0};
38,584✔
57
    taosGetTmpfilePath(pBuf->prefix, "paged-buf", path);
38,584✔
58
    pBuf->path = taosStrdup(path);
38,584!
59
    if (pBuf->path == NULL) {
38,584!
60
      return terrno;
×
61
    }
62
  }
63

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

70
  return TSDB_CODE_SUCCESS;
38,584✔
71
}
72

73
static char* doCompressData(void* data, int32_t srcSize, int32_t* dst, SDiskbasedBuf* pBuf) {  // do nothing
167,754,530✔
74
  if (!pBuf->comp) {
167,754,530!
75
    *dst = srcSize;
167,754,530✔
76
    return data;
167,754,530✔
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
143,835,003✔
86
  int32_t code = 0;
143,835,003✔
87
  if (!pBuf->comp) {
143,835,003!
88
    *dst = srcSize;
143,835,227✔
89
    return code;
143,835,227✔
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) {
157,116,854✔
103
  if (pBuf->pFree == NULL) {
157,116,854!
104
    return pBuf->nextPos;
×
105
  } else {
106
    int32_t offset = -1;
157,116,854✔
107

108
    size_t num = taosArrayGetSize(pBuf->pFree);
157,116,854✔
109
    for (int32_t i = 0; i < num; ++i) {
157,116,854!
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;
157,116,854✔
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); }
686,935,229✔
135

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

142
  ret = (int32_t)taosWriteFile(pBuf->pFile, pData, size);
167,754,306✔
143
  if (ret != size) {
167,754,306!
144
    return terrno;
×
145
  }
146

147
  // extend the file
148
  if (pBuf->fileSize < offset + size) {
167,754,306✔
149
    pBuf->fileSize = offset + size;
157,116,854✔
150
  }
151

152
  pBuf->statis.flushBytes += size;
167,754,530✔
153
  pBuf->statis.flushPages += 1;
167,754,530✔
154

155
  return TSDB_CODE_SUCCESS;
167,754,530✔
156
}
157

158
static char* doFlushBufPage(SDiskbasedBuf* pBuf, SPageInfo* pg) {
244,039,992✔
159
  if (pg->pData == NULL || pg->used) {
244,039,992!
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;
244,039,992✔
166
  int64_t offset = pg->offset;
244,039,992✔
167

168
  char* t = NULL;
244,039,992✔
169
  if ((!HAS_DATA_IN_DISK(pg)) || pg->dirty) {
244,039,992✔
170
    void* payload = GET_PAYLOAD_DATA(pg);
167,754,530✔
171
    t = doCompressData(payload, pBuf->pageSize + sizeof(SFilePage), &size, pBuf);
167,754,530✔
172
    if (size < 0) {
167,754,530!
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) {
244,039,992✔
181
    if (!HAS_DATA_IN_DISK(pg)) {
167,754,530✔
182
      offset = allocateNewPositionInFile(pBuf, size);
157,116,854✔
183
      pBuf->nextPos += size;
157,116,854✔
184

185
      int32_t code = doFlushBufPageImpl(pBuf, offset, t, size);
157,116,796✔
186
      if (code != TSDB_CODE_SUCCESS) {
157,116,854!
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) {
10,637,676!
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);
10,637,676✔
204
      if (code != TSDB_CODE_SUCCESS) {
10,637,676!
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;
76,285,462✔
210
  }
211

212
  char* pDataBuf = pg->pData;
244,039,992✔
213
  memset(pDataBuf, 0, getAllocPageSize(pBuf->pageSize));
488,079,760!
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;
244,039,992✔
220
  pg->length = size;  // on disk size
244,039,992✔
221
  return pDataBuf;
244,039,992✔
222
}
223

224
static char* flushBufPage(SDiskbasedBuf* pBuf, SPageInfo* pg) {
244,039,956✔
225
  int32_t ret = TSDB_CODE_SUCCESS;
244,039,956✔
226

227
  if (pBuf->pFile == NULL) {
244,039,956✔
228
    if ((ret = createDiskFile(pBuf)) != TSDB_CODE_SUCCESS) {
38,584!
229
      terrno = ret;
×
230
      return NULL;
×
231
    }
232
  }
233

234
  char* p = doFlushBufPage(pBuf, pg);
244,039,956✔
235
  CLEAR_BUF_PAGE_IN_MEM_FLAG(pg);
244,039,992✔
236

237
  pg->dirty = false;
244,039,992✔
238
  return p;
244,039,992✔
239
}
240

241
// load file block data in disk
242
static int32_t loadPageFromDisk(SDiskbasedBuf* pBuf, SPageInfo* pg) {
143,835,003✔
243
  if (pg->offset < 0 || pg->length <= 0) {
143,835,003!
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);
143,835,227✔
249
  if (ret < 0) {
143,835,003!
250
    ret = terrno;
×
251
    return ret;
×
252
  }
253

254
  void* pPage = (void*)GET_PAYLOAD_DATA(pg);
143,835,003✔
255
  ret = taosReadFile(pBuf->pFile, pPage, pg->length);
143,834,779✔
256
  if (ret != pg->length) {
143,835,003!
257
    ret = terrno;
×
258
    return ret;
×
259
  }
260

261
  pBuf->statis.loadBytes += pg->length;
143,835,227✔
262
  pBuf->statis.loadPages += 1;
143,835,227✔
263

264
  int32_t fullSize = 0;
143,835,003✔
265
  return doDecompressData(pPage, pg->length, &fullSize, pBuf);
143,835,003✔
266
}
267

268
static SPageInfo* registerNewPageInfo(SDiskbasedBuf* pBuf, int32_t pageId) {
543,076,712✔
269
  pBuf->numOfPages += 1;
543,076,712✔
270

271
  SPageInfo* ppi = taosMemoryMalloc(sizeof(SPageInfo));
543,103,924!
272
  if (ppi == NULL) {
543,005,588!
273
    return NULL;
×
274
  }
275

276
  ppi->pageId = pageId;
543,005,588✔
277
  ppi->pData = NULL;
543,038,997✔
278
  ppi->offset = -1;
543,058,125✔
279
  ppi->length = -1;
543,037,041✔
280
  ppi->used = true;
543,049,346✔
281
  ppi->pn = NULL;
543,066,767✔
282
  ppi->dirty = false;
543,063,283✔
283

284
  SPageInfo** pRet = taosArrayPush(pBuf->pIdList, &ppi);
543,076,723✔
285
  if (NULL == pRet) {
543,041,605!
286
    taosMemoryFree(ppi);
×
287
    return NULL;
×
288
  }
289
  return *pRet;
543,041,605✔
290
}
291

292
static SListNode* getEldestUnrefedPage(SDiskbasedBuf* pBuf) {
244,039,992✔
293
  SListIter iter = {0};
244,039,992✔
294
  tdListInitIter(pBuf->lruList, &iter, TD_LIST_BACKWARD);
244,039,992✔
295

296
  SListNode* pn = NULL;
244,039,934✔
297
  while ((pn = tdListNext(&iter)) != NULL) {
263,822,439!
298
    SPageInfo* pageInfo = *(SPageInfo**)pn->data;
263,822,215✔
299

300
    SPageInfo* p = *(SPageInfo**)(pageInfo->pData);
263,822,215✔
301

302
    if (!pageInfo->used) {
263,822,215✔
303
      break;
244,039,992✔
304
    }
305
  }
306

307
  return pn;
244,039,992✔
308
}
309

310
static char* evictBufPage(SDiskbasedBuf* pBuf) {
244,039,992✔
311
  SListNode* pn = getEldestUnrefedPage(pBuf);
244,039,992✔
312
  if (pn == NULL) {  // no available buffer pages now, return.
244,039,934!
313
    return NULL;
×
314
  }
315

316
  terrno = 0;
244,039,934✔
317
  pn = tdListPopNode(pBuf->lruList, pn);
244,039,934✔
318

319
  SPageInfo* d = *(SPageInfo**)pn->data;
244,039,992✔
320

321
  d->pn = NULL;
244,039,992✔
322
  taosMemoryFreeClear(pn);
244,039,992!
323

324
  return flushBufPage(pBuf, d);
244,039,956✔
325
}
326

327
static int32_t lruListPushFront(SList* pList, SPageInfo* pi) {
686,841,376✔
328
  int32_t code = tdListPrepend(pList, &pi);
686,841,376✔
329
  if (TSDB_CODE_SUCCESS != code) {
686,942,466!
330
    return code;
×
331
  }
332
  SListNode* front = tdListGetHead(pList);
686,942,466✔
333
  pi->pn = front;
686,899,731✔
334
  return TSDB_CODE_SUCCESS;
686,847,590✔
335
}
336

337
static void lruListMoveToFront(SList* pList, SPageInfo* pi) {
2,147,483,647✔
338
  pi->pn = tdListPopNode(pList, pi->pn);
2,147,483,647✔
339
  tdListPrependNode(pList, pi->pn);
2,147,483,647✔
340
}
2,147,483,647✔
341

342
static SPageInfo* getPageInfoFromPayload(void* page) {
2,147,483,647✔
343
  char* p = (char*)page - POINTER_BYTES;
2,147,483,647✔
344

345
  SPageInfo* ppi = ((SPageInfo**)p)[0];
2,147,483,647✔
346
  return ppi;
2,147,483,647✔
347
}
348

349
int32_t createDiskbasedBuf(SDiskbasedBuf** pBuf, int32_t pagesize, int64_t inMemBufSize, const char* id,
55,513,786✔
350
                           const char* dir) {
351
  int32_t code = 0;
55,513,786✔
352
  *pBuf = NULL;
55,513,786✔
353
  SDiskbasedBuf* pPBuf = taosMemoryCalloc(1, sizeof(SDiskbasedBuf));
55,526,765!
354
  if (pPBuf == NULL) {
55,453,639!
355
    code = terrno;
×
356
    goto _error;
×
357
  }
358

359
  pPBuf->pageSize = pagesize;
55,453,639✔
360
  pPBuf->numOfPages = 0;  // all pages are in buffer in the first place
55,458,707✔
361
  pPBuf->totalBufSize = 0;
55,460,780✔
362
  pPBuf->allocateId = -1;
55,468,337✔
363
  pPBuf->pFile = NULL;
55,494,924✔
364
  pPBuf->id = taosStrdup(id);
55,510,319!
365
  if (id != NULL && pPBuf->id == NULL) {
55,526,456!
366
    code = terrno;
×
367
    goto _error;
×
368
  }
369
  pPBuf->fileSize = 0;
55,524,286✔
370
  pPBuf->pFree = taosArrayInit(4, sizeof(SFreeListItem));
55,526,367✔
371
  pPBuf->freePgList = tdListNew(POINTER_BYTES);
55,502,358✔
372
  if (pPBuf->pFree == NULL || pPBuf->freePgList == NULL) {
55,493,819✔
373
    code = terrno;
13,819✔
374
    goto _error;
×
375
  }
376

377
  // at least more than 2 pages must be in memory
378
  if (inMemBufSize < pagesize * 2) {
55,497,088✔
379
    inMemBufSize = pagesize * 2;
447,840✔
380
  }
381

382
  pPBuf->inMemPages = inMemBufSize / pagesize;  // maximum allowed pages, it is a soft limit.
55,497,088!
383
  pPBuf->lruList = tdListNew(POINTER_BYTES);
55,513,061✔
384
  if (pPBuf->lruList == NULL) {
55,522,340!
385
    code = terrno;
×
386
    goto _error;
×
387
  }
388

389
  // init id hash table
390
  _hash_fn_t fn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT);
55,513,284✔
391
  pPBuf->pIdList = taosArrayInit(4, POINTER_BYTES);
55,515,813✔
392
  if (pPBuf->pIdList == NULL) {
55,517,704!
393
    code = terrno;
×
394
    goto _error;
×
395
  }
396

397
  pPBuf->all = tSimpleHashInit(64, fn);
55,506,874✔
398
  if (pPBuf->all == NULL) {
55,502,400!
399
    code = terrno;
×
400
    goto _error;
×
401
  }
402

403
  pPBuf->prefix = (char*)dir;
55,499,811✔
404
  pPBuf->emptyDummyIdList = taosArrayInit(1, sizeof(int32_t));
55,506,791✔
405
  if (pPBuf->emptyDummyIdList == NULL) {
55,521,721!
406
    code = terrno;
×
407
    goto _error;
×
408
  }
409

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

413
  *pBuf = pPBuf;
55,515,977✔
414
  return TSDB_CODE_SUCCESS;
55,514,923✔
415

416
_error:
×
417
  destroyDiskbasedBuf(pPBuf);
×
418
  *pBuf = NULL;
×
419
  return code;
×
420
}
421

422
static char* doExtractPage(SDiskbasedBuf* pBuf) {
686,833,493✔
423
  char* availablePage = NULL;
686,833,493✔
424
  if (NO_IN_MEM_AVAILABLE_PAGES(pBuf)) {
686,833,493✔
425
    availablePage = evictBufPage(pBuf);
244,039,992✔
426
    if (availablePage == NULL) {
244,039,992!
427
      uWarn("no available buf pages, current:%d, max:%d, reason: %s, %s", listNEles(pBuf->lruList), pBuf->inMemPages,
×
428
            terrstr(), pBuf->id)
429
    }
430
  } else {
431
    availablePage =
442,526,525✔
432
        taosMemoryCalloc(1, getAllocPageSize(pBuf->pageSize));  // add extract bytes in case of zipped buffer increased.
885,749,765!
433
  }
434

435
  return availablePage;
686,566,618✔
436
}
437

438
void* getNewBufPage(SDiskbasedBuf* pBuf, int32_t* pageId) {
542,969,245✔
439
  pBuf->statis.getPages += 1;
542,969,245✔
440

441
  char* availablePage = doExtractPage(pBuf);
543,071,101✔
442
  if (availablePage == NULL) {
542,731,867!
443
    return NULL;
×
444
  }
445

446
  SPageInfo* pi = NULL;
542,731,867✔
447
  int32_t    code = 0;
542,738,258✔
448
  if (listNEles(pBuf->freePgList) != 0) {
542,738,258!
449
    SListNode* pItem = tdListPopHead(pBuf->freePgList);
×
450
    pi = *(SPageInfo**)pItem->data;
×
451
    pi->used = true;
×
452
    *pageId = pi->pageId;
×
453
    taosMemoryFreeClear(pItem);
×
454
    code = lruListPushFront(pBuf->lruList, pi);
×
455
    if (TSDB_CODE_SUCCESS != code) {
×
456
      taosMemoryFree(pi);
×
457
      taosMemoryFree(availablePage);
×
458
      terrno = code;
×
459
      return NULL;
×
460
    }
461
  } else {  // create a new pageinfo
462
    // register new id in this group
463
    *pageId = (++pBuf->allocateId);
542,841,010✔
464

465
    // register page id info
466
    pi = registerNewPageInfo(pBuf, *pageId);
542,960,747✔
467
    if (pi == NULL) {
543,057,954!
468
      taosMemoryFree(availablePage);
×
469
      return NULL;
×
470
    }
471

472
    // add to hash map
473
    int32_t code = tSimpleHashPut(pBuf->all, pageId, sizeof(int32_t), &pi, POINTER_BYTES);
543,057,954✔
474

475
    if (TSDB_CODE_SUCCESS == code) {
543,046,918✔
476
      // add to LRU list
477
      code = lruListPushFront(pBuf->lruList, pi);
543,015,035✔
478
    }
479
    if (TSDB_CODE_SUCCESS == code) {
543,042,988!
480
      pBuf->totalBufSize += pBuf->pageSize;
543,042,988✔
481
    } else {
482
      taosMemoryFree(availablePage);
×
483
      SPageInfo **pLast = taosArrayPop(pBuf->pIdList);
×
484
      int32_t ret = tSimpleHashRemove(pBuf->all, pageId, sizeof(int32_t));
×
485
      if (ret != TSDB_CODE_SUCCESS) {
×
486
        uError("%s failed to clear pageId %d from buf hash-set since %s", __func__, *pageId, tstrerror(ret));
×
487
      }
488
      taosMemoryFree(pi);
×
489
      terrno = code;
×
490
      return NULL;
×
491
    }
492
  }
493

494
  pi->pData = availablePage;
543,090,506✔
495

496
  ((void**)pi->pData)[0] = pi;
543,051,619✔
497
#ifdef BUF_PAGE_DEBUG
498
  uDebug("page_getNewBufPage , pi->pData:%p, pageId:%d, offset:%" PRId64, pi->pData, pi->pageId, pi->offset);
499
#endif
500

501
  return (void*)(GET_PAYLOAD_DATA(pi));
543,068,645✔
502
}
503

504
void* getBufPage(SDiskbasedBuf* pBuf, int32_t id) {
2,147,483,647✔
505
  if (id < 0) {
2,147,483,647!
506
    terrno = TSDB_CODE_INVALID_PARA;
×
507
    uError("invalid page id:%d, %s", id, pBuf->id);
×
508
    return NULL;
×
509
  }
510

511
  pBuf->statis.getPages += 1;
2,147,483,647✔
512

513
  SPageInfo** pi = tSimpleHashGet(pBuf->all, &id, sizeof(int32_t));
2,147,483,647✔
514
  if (pi == NULL || *pi == NULL) {
2,147,483,647!
515
    uError("failed to locate the buffer page:%d, %s", id, pBuf->id);
400!
516
    terrno = TSDB_CODE_INVALID_PARA;
400✔
517
    return NULL;
×
518
  }
519

520
  if (BUF_PAGE_IN_MEM(*pi)) {  // it is in memory
2,147,483,647✔
521
    // no need to update the LRU list if only one page exists
522
    if (pBuf->numOfPages == 1) {
2,147,483,647✔
523
      (*pi)->used = true;
1,304,898,178✔
524
      return (void*)(GET_PAYLOAD_DATA(*pi));
1,304,988,001✔
525
    }
526

527
    SPageInfo** pInfo = (SPageInfo**)((*pi)->pn->data);
2,147,483,647✔
528
    if (*pInfo != *pi) {
2,147,483,647!
529
      terrno = TSDB_CODE_APP_ERROR;
×
530
      uError("inconsistently data in paged buffer, pInfo:%p, pi:%p, %s", *pInfo, *pi, pBuf->id);
×
531
      return NULL;
×
532
    }
533

534
    lruListMoveToFront(pBuf->lruList, (*pi));
2,147,483,647✔
535
    (*pi)->used = true;
2,147,483,647✔
536

537
#ifdef BUF_PAGE_DEBUG
538
    uDebug("page_getBufPage1 pageId:%d, offset:%" PRId64, (*pi)->pageId, (*pi)->offset);
539
#endif
540
    return (void*)(GET_PAYLOAD_DATA(*pi));
2,147,483,647✔
541
  } else {  // not in memory
542

543
    (*pi)->pData = doExtractPage(pBuf);
143,835,227✔
544

545
    // failed to evict buffer page, return with error code.
546
    if ((*pi)->pData == NULL) {
143,835,227!
547
      return NULL;
×
548
    }
549

550
    // set the ptr to the new SPageInfo
551
    ((void**)((*pi)->pData))[0] = (*pi);
143,835,227✔
552

553
    int32_t code = lruListPushFront(pBuf->lruList, *pi);
143,835,227✔
554
    if (TSDB_CODE_SUCCESS != code) {
143,835,227!
555
      taosMemoryFree((*pi)->pData);
×
556
      (*pi)->pData = NULL;
×
557
      terrno = code;
×
558
      return NULL;
×
559
    }
560
    (*pi)->used = true;
143,835,227✔
561

562
    // some data has been flushed to disk, and needs to be loaded into buffer again.
563
    if (HAS_DATA_IN_DISK(*pi)) {
143,835,227!
564
      int32_t code = loadPageFromDisk(pBuf, *pi);
143,835,227✔
565
      if (code != 0) {
143,835,227!
566
        taosMemoryFree((*pi)->pData);
×
567
        (*pi)->pData = NULL;
×
568
        terrno = code;
×
569
        return NULL;
×
570
      }
571
    }
572
#ifdef BUF_PAGE_DEBUG
573
    uDebug("page_getBufPage2 pageId:%d, offset:%" PRId64, (*pi)->pageId, (*pi)->offset);
574
#endif
575
    return (void*)(GET_PAYLOAD_DATA(*pi));
143,835,227✔
576
  }
577
}
578

579
void releaseBufPage(SDiskbasedBuf* pBuf, void* page) {
2,147,483,647✔
580
  if (page == NULL) {
2,147,483,647!
581
    return;
×
582
  }
583

584
  SPageInfo* ppi = getPageInfoFromPayload(page);
2,147,483,647✔
585
  releaseBufPageInfo(pBuf, ppi);
2,147,483,647✔
586
}
587

588
void releaseBufPageInfo(SDiskbasedBuf* pBuf, SPageInfo* pi) {
2,147,483,647✔
589
#ifdef BUF_PAGE_DEBUG
590
  uDebug("page_releaseBufPageInfo pageId:%d, used:%d, offset:%" PRId64, pi->pageId, pi->used, pi->offset);
591
#endif
592

593
  if (pi == NULL) {
2,147,483,647!
594
    return;
×
595
  }
596

597
  if (pi->pData == NULL) {
2,147,483,647!
598
    uError("pi->pData (page data) is null");
×
599
    return;
×
600
  }
601

602
  pi->used = false;
2,147,483,647✔
603
  pBuf->statis.releasePages += 1;
2,147,483,647✔
604
}
605

UNCOV
606
size_t getTotalBufSize(const SDiskbasedBuf* pBuf) { return (size_t)pBuf->totalBufSize; }
×
607

UNCOV
608
SArray* getDataBufPagesIdList(SDiskbasedBuf* pBuf) { return pBuf->pIdList; }
×
609

610
void destroyDiskbasedBuf(SDiskbasedBuf* pBuf) {
63,512,855✔
611
  if (pBuf == NULL) {
63,512,855✔
612
    return;
7,972,273✔
613
  }
614

615
  dBufPrintStatis(pBuf);
55,540,582✔
616

617
  bool needRemoveFile = false;
55,540,199✔
618
  if (pBuf->pFile != NULL) {
55,540,199✔
619
    needRemoveFile = true;
38,584✔
620
    uDebug(
38,584✔
621
        "Paged buffer closed, total:%.2f Kb (%d Pages), inmem size:%.2f Kb (%d Pages), file size:%.2f Kb, page "
622
        "size:%.2f Kb, %s",
623
        pBuf->totalBufSize / 1024.0, pBuf->numOfPages, listNEles(pBuf->lruList) * pBuf->pageSize / 1024.0,
624
        listNEles(pBuf->lruList), pBuf->fileSize / 1024.0, pBuf->pageSize / 1024.0f, pBuf->id);
625

626
    int32_t code = taosCloseFile(&pBuf->pFile);
38,584✔
627
    if (TSDB_CODE_SUCCESS != code) {
38,584!
628
      uDebug("WARNING tPage failed to close file when destroy disk basebuf: %s", pBuf->path);
×
629
    }
630
  } else {
631
    uDebug("Paged buffer closed, total:%.2f Kb, no file created, %s", pBuf->totalBufSize / 1024.0, pBuf->id);
55,502,303✔
632
  }
633

634
  // print the statistics information
635
  {
636
    SDiskbasedBufStatis* ps = &pBuf->statis;
55,540,887✔
637
    if (ps->loadPages == 0) {
55,543,149✔
638
      uDebug("Get/Release pages:%d/%d, flushToDisk:%.2f Kb (%d Pages), loadFromDisk:%.2f Kb (%d Pages)", ps->getPages,
55,504,563✔
639
             ps->releasePages, ps->flushBytes / 1024.0f, ps->flushPages, ps->loadBytes / 1024.0f, ps->loadPages);
640
    } else {
641
      uDebug(
38,584!
642
          "Get/Release pages:%d/%d, flushToDisk:%.2f Kb (%d Pages), loadFromDisk:%.2f Kb (%d Pages), avgPgSize:%.2f Kb",
643
          ps->getPages, ps->releasePages, ps->flushBytes / 1024.0f, ps->flushPages, ps->loadBytes / 1024.0f,
644
          ps->loadPages, ps->loadBytes / (1024.0 * ps->loadPages));
645
    }
646
  }
647

648
  if (needRemoveFile) {
55,542,575✔
649
    int32_t ret = taosRemoveFile(pBuf->path);
38,584✔
650
    if (ret != 0) {  // print the error and discard this error info
38,584!
651
      uDebug("WARNING tPage remove file failed. path=%s, code:%s", pBuf->path, strerror(ERRNO));
38,584✔
652
    }
653
  }
654

655
  taosMemoryFreeClear(pBuf->path);
55,542,575!
656

657
  size_t n = taosArrayGetSize(pBuf->pIdList);
55,543,580✔
658
  for (int32_t i = 0; i < n; ++i) {
487,995,058✔
659
    SPageInfo* pi = taosArrayGetP(pBuf->pIdList, i);
432,453,391✔
660
    taosMemoryFreeClear(pi->pData);
432,437,629!
661
    taosMemoryFreeClear(pi);
432,407,610!
662
  }
663

664
  taosArrayDestroy(pBuf->pIdList);
55,541,667✔
665

666
  pBuf->lruList = tdListFree(pBuf->lruList);
55,540,628✔
667
  pBuf->freePgList = tdListFree(pBuf->freePgList);
55,539,259✔
668

669
  taosArrayDestroy(pBuf->emptyDummyIdList);
55,539,544✔
670
  taosArrayDestroy(pBuf->pFree);
55,539,576✔
671

672
  tSimpleHashCleanup(pBuf->all);
55,539,453✔
673

674
  taosMemoryFreeClear(pBuf->id);
55,541,784!
675
  taosMemoryFreeClear(pBuf->assistBuf);
55,538,558!
676
  taosMemoryFreeClear(pBuf);
55,541,480!
677
}
678

679
SPageInfo* getLastPageInfo(SArray* pList) {
×
680
  size_t     size = taosArrayGetSize(pList);
×
681
  SPageInfo* pPgInfo = taosArrayGetP(pList, size - 1);
×
682
  return pPgInfo;
×
683
}
684

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

687
int32_t getBufPageSize(const SDiskbasedBuf* pBuf) { return pBuf->pageSize; }
2,147,483,647✔
688

689
int32_t getNumOfInMemBufPages(const SDiskbasedBuf* pBuf) { return pBuf->inMemPages; }
2,286,415✔
690

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

693
void setBufPageDirty(void* pPage, bool dirty) {
2,147,483,647✔
694
  SPageInfo* ppi = getPageInfoFromPayload(pPage);
2,147,483,647✔
695
  ppi->dirty = dirty;
2,147,483,647✔
696
}
2,147,483,647✔
697

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

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

712
  int32_t code = tdListAppend(pBuf->freePgList, &ppi);
×
713
  if (TSDB_CODE_SUCCESS != code) {
×
714
    return code;
×
715
  }
716

717
  ppi->used = false;
×
718
  ppi->dirty = false;
×
719

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

728
void dBufSetPrintInfo(SDiskbasedBuf* pBuf) { pBuf->printStatis = true; }
7,385,867✔
729

730
SDiskbasedBufStatis getDBufStatis(const SDiskbasedBuf* pBuf) { return pBuf->statis; }
2,277,269✔
731

732
void dBufPrintStatis(const SDiskbasedBuf* pBuf) {
55,542,266✔
733
  if (!pBuf->printStatis) {
55,542,266!
734
    return;
48,155,445✔
735
  }
736

737
  const SDiskbasedBufStatis* ps = &pBuf->statis;
7,387,181✔
738

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

747
  if (ps->loadPages > 0) {
7,387,197✔
748
    (void)printf(
9,594✔
749
        "Get/Release pages:%d/%d, flushToDisk:%.2f Kb (%d Pages), loadFromDisk:%.2f Kb (%d Pages), avgPageSize:%.2f "
750
        "Kb\n",
751
        ps->getPages, ps->releasePages, ps->flushBytes / 1024.0f, ps->flushPages, ps->loadBytes / 1024.0f,
9,594!
752
        ps->loadPages, ps->loadBytes / (1024.0 * ps->loadPages));
9,594!
753
  } else {
754
    // printf("no page loaded\n");
755
  }
756
}
757

758
void clearDiskbasedBuf(SDiskbasedBuf* pBuf) {
447,273✔
759
  size_t n = taosArrayGetSize(pBuf->pIdList);
447,273✔
760
  for (int32_t i = 0; i < n; ++i) {
111,150,599✔
761
    SPageInfo* pi = taosArrayGetP(pBuf->pIdList, i);
110,703,326✔
762
    taosMemoryFreeClear(pi->pData);
110,703,326!
763
    taosMemoryFreeClear(pi);
110,702,735!
764
  }
765

766
  taosArrayClear(pBuf->pIdList);
447,273✔
767

768
  tdListEmpty(pBuf->lruList);
447,273✔
769
  tdListEmpty(pBuf->freePgList);
447,273✔
770

771
  taosArrayClear(pBuf->emptyDummyIdList);
447,273✔
772
  taosArrayClear(pBuf->pFree);
447,273✔
773

774
  tSimpleHashClear(pBuf->all);
447,273✔
775

776
  pBuf->numOfPages = 0;  // all pages are in buffer in the first place
447,273✔
777
  pBuf->totalBufSize = 0;
447,273✔
778
  pBuf->allocateId = -1;
447,273✔
779
  pBuf->fileSize = 0;
447,273✔
780
}
447,273✔
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