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

taosdata / TDengine / #5023

15 Apr 2026 11:15AM UTC coverage: 72.251% (-0.03%) from 72.278%
#5023

push

travis-ci

web-flow
feat: implement fixed bucket table distribution (#35092)

221 of 236 new or added lines in 4 files covered. (93.64%)

613 existing lines in 123 files now uncovered.

257631 of 356577 relevant lines covered (72.25%)

133876647.56 hits per line

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

69.47
/source/dnode/vnode/src/vnd/vnodeBufPool.c
1
/*
2
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3
 *
4
 * This program is free software: you can use, redistribute, and/or modify
5
 * it under the terms of the GNU Affero General Public License, version 3
6
 * or later ("AGPL"), as published by the Free Software Foundation.
7
 *
8
 * This program is distributed in the hope that it will be useful, but WITHOUT
9
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10
 * FITNESS FOR A PARTICULAR PURPOSE.
11
 *
12
 * You should have received a copy of the GNU Affero General Public License
13
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
14
 */
15

16
#include "vnd.h"
17

18
/* ------------------------ STRUCTURES ------------------------ */
19
static int32_t vnodeBufPoolCreate(SVnode *pVnode, int32_t id, int64_t size, SVBufPool **ppPool) {
12,968,118✔
20
  SVBufPool *pPool;
21

22
  pPool = taosMemoryCalloc(1, sizeof(SVBufPool));
12,968,118✔
23
  if (pPool == NULL) {
12,968,422✔
24
    vError("vgId:%d, failed to allocate buffer pool", TD_VID(pVnode));
×
25
    return terrno;
×
26
  }
27
  pPool->node.data = taosMemoryMalloc(size);
12,968,422✔
28
  if (NULL == pPool->node.data) {
12,970,956✔
29
    vError("vgId:%d, failed to allocate buffer pool data, size:%" PRId64, TD_VID(pVnode), size);
×
30
    taosMemoryFree(pPool);
×
31
    return terrno;
×
32
  }
33

34
  // query handle list
35
  (void)taosThreadMutexInit(&pPool->mutex, NULL);
12,970,858✔
36
  pPool->nQuery = 0;
12,970,374✔
37
  pPool->qList.pNext = &pPool->qList;
12,970,374✔
38
  pPool->qList.ppNext = &pPool->qList.pNext;
12,970,057✔
39

40
  pPool->pVnode = pVnode;
12,969,836✔
41
  pPool->id = id;
12,969,563✔
42
  pPool->ptr = pPool->node.data;
12,970,428✔
43
  pPool->pTail = &pPool->node;
12,970,003✔
44
  pPool->node.prev = NULL;
12,970,639✔
45
  pPool->node.pnext = &pPool->pTail;
12,970,858✔
46
  pPool->node.size = size;
12,970,541✔
47

48
  *ppPool = pPool;
12,970,541✔
49
  return 0;
12,969,309✔
50
}
51

52
static void vnodeBufPoolDestroy(SVBufPool *pPool) {
12,970,864✔
53
  vnodeBufPoolReset(pPool);
12,970,864✔
54
  (void)taosThreadMutexDestroy(&pPool->mutex);
12,970,956✔
55
  taosMemoryFree(pPool->node.data);
12,970,956✔
56
  taosMemoryFree(pPool);
12,970,956✔
57
}
12,970,956✔
58

59
int vnodeOpenBufPool(SVnode *pVnode) {
4,321,561✔
60
  int64_t size = pVnode->config.szBuf / VNODE_BUFPOOL_SEGMENTS;
4,321,561✔
61

62
  for (int i = 0; i < VNODE_BUFPOOL_SEGMENTS; i++) {
17,291,114✔
63
    // create pool
64
    int32_t code;
65
    if ((code = vnodeBufPoolCreate(pVnode, i, size, &pVnode->aBufPool[i]))) {
12,967,956✔
66
      vError("vgId:%d, failed to open vnode buffer pool since %s", TD_VID(pVnode), tstrerror(terrno));
×
67
      vnodeCloseBufPool(pVnode);
×
68
      return code;
×
69
    }
70

71
    // add to free list
72
    pVnode->aBufPool[i]->freeNext = pVnode->freeList;
12,969,309✔
73
    pVnode->freeList = pVnode->aBufPool[i];
12,969,498✔
74
  }
75

76
  vDebug("vgId:%d, vnode buffer pool is opened, size:%" PRId64, TD_VID(pVnode), size);
4,323,158✔
77
  return 0;
4,323,652✔
78
}
79

80
void vnodeCloseBufPool(SVnode *pVnode) {
4,323,652✔
81
  for (int32_t i = 0; i < VNODE_BUFPOOL_SEGMENTS; i++) {
17,294,608✔
82
    if (pVnode->aBufPool[i]) {
12,970,956✔
83
      vnodeBufPoolDestroy(pVnode->aBufPool[i]);
12,970,956✔
84
      pVnode->aBufPool[i] = NULL;
12,970,956✔
85
    }
86
  }
87

88
  vDebug("vgId:%d, vnode buffer pool is closed", TD_VID(pVnode));
4,323,652✔
89
}
4,323,652✔
90

91
void vnodeBufPoolReset(SVBufPool *pPool) {
18,400,180✔
92
  if (pPool->nQuery != 0) {
18,400,180✔
93
    vError("vgId:%d, buffer pool %p of id %d has %d queries, reset it may cause problem", TD_VID(pPool->pVnode), pPool,
×
94
           pPool->id, pPool->nQuery);
95
  }
96

97
  for (SVBufPoolNode *pNode = pPool->pTail; pNode->prev; pNode = pPool->pTail) {
1,051,468,487✔
98
    pNode->prev->pnext = &pPool->pTail;
1,033,068,307✔
99
    pPool->pTail = pNode->prev;
1,033,068,307✔
100
    pPool->size = pPool->size - sizeof(*pNode) - pNode->size;
1,033,068,307✔
101
    taosMemoryFree(pNode);
1,033,068,307✔
102
  }
103

104
  pPool->size = 0;
18,400,180✔
105
  pPool->ptr = pPool->node.data;
18,400,180✔
106
}
18,400,180✔
107

108
void *vnodeBufPoolMallocAligned(SVBufPool *pPool, int size) {
2,147,483,647✔
109
  SVBufPoolNode *pNode;
110
  void          *p = NULL;
2,147,483,647✔
111
  uint8_t       *ptr = NULL;
2,147,483,647✔
112
  int            paddingLen = 0;
2,147,483,647✔
113

114
  if (pPool == NULL) {
2,147,483,647✔
115
    terrno = TSDB_CODE_INVALID_PARA;
×
116
    return NULL;
×
117
  }
118

119
  ptr = pPool->ptr;
2,147,483,647✔
120
  paddingLen = (((long)ptr + 7) & ~7) - (long)ptr;
2,147,483,647✔
121

122
  if (pPool->node.size >= pPool->ptr - pPool->node.data + size + paddingLen) {
2,147,483,647✔
123
    // allocate from the anchor node
124
    p = pPool->ptr + paddingLen;
2,147,483,647✔
125
    size += paddingLen;
2,147,483,647✔
126
    pPool->ptr = pPool->ptr + size;
2,147,483,647✔
127
    pPool->size += size;
2,147,483,647✔
128
  } else {
129
    // allocate a new node
130
    pNode = taosMemoryMalloc(sizeof(*pNode) + size);
1,033,068,307✔
131
    if (pNode == NULL) {
1,033,068,307✔
132
      return NULL;
×
133
    }
134
    pNode->data = (uint8_t *)&pNode[1];
1,033,068,307✔
135

136
    p = pNode->data;
1,033,068,307✔
137
    pNode->size = size;
1,033,068,307✔
138
    pNode->prev = pPool->pTail;
1,033,068,307✔
139
    pNode->pnext = &pPool->pTail;
1,033,068,307✔
140
    pPool->pTail->pnext = &pNode->prev;
1,033,068,307✔
141
    pPool->pTail = pNode;
1,033,068,307✔
142

143
    pPool->size = pPool->size + sizeof(*pNode) + size;
1,033,068,307✔
144
  }
145
  return p;
2,147,483,647✔
146
}
147

148
void *vnodeBufPoolMalloc(SVBufPool *pPool, int size) {
44,346,232✔
149
  SVBufPoolNode *pNode;
150
  void          *p = NULL;
44,346,232✔
151

152
  if (pPool == NULL) {
44,346,232✔
153
    terrno = TSDB_CODE_INVALID_PARA;
×
154
    return NULL;
×
155
  }
156

157
  if (pPool->node.size >= pPool->ptr - pPool->node.data + size) {
44,346,232✔
158
    // allocate from the anchor node
159
    p = pPool->ptr;
44,346,232✔
160
    pPool->ptr = pPool->ptr + size;
44,346,232✔
161
    pPool->size += size;
44,346,232✔
162
  } else {
163
    // allocate a new node
164
    pNode = taosMemoryMalloc(sizeof(*pNode) + size);
×
165
    if (pNode == NULL) {
×
166
      return NULL;
×
167
    }
168
    pNode->data = (uint8_t *)&pNode[1];
×
169

170
    p = pNode->data;
×
171
    pNode->size = size;
×
172
    pNode->prev = pPool->pTail;
×
173
    pNode->pnext = &pPool->pTail;
×
174
    pPool->pTail->pnext = &pNode->prev;
×
175
    pPool->pTail = pNode;
×
176

177
    pPool->size = pPool->size + sizeof(*pNode) + size;
×
178
  }
179
  return p;
44,346,232✔
180
}
181

182
void vnodeBufPoolFree(SVBufPool *pPool, void *p) {
122,878,203✔
183
  // uint8_t       *ptr = (uint8_t *)p;
184
  // SVBufPoolNode *pNode;
185

186
  // if (ptr < pPool->node.data || ptr >= pPool->node.data + pPool->node.size) {
187
  //   pNode = &((SVBufPoolNode *)p)[-1];
188
  //   *pNode->pnext = pNode->prev;
189
  //   pNode->prev->pnext = pNode->pnext;
190

191
  //   pPool->size = pPool->size - sizeof(*pNode) - pNode->size;
192
  //   taosMemoryFree(pNode);
193
  // }
194
}
122,878,203✔
195

196
void vnodeBufPoolRef(SVBufPool *pPool) {
9,750,256✔
197
  int32_t nRef = atomic_fetch_add_32(&pPool->nRef, 1);
9,750,256✔
198
  if (nRef <= 0) {
9,751,495✔
199
    vError("vgId:%d, buffer pool %p of id %d is referenced by %d", TD_VID(pPool->pVnode), pPool, pPool->id, nRef);
×
200
  }
201
}
9,751,495✔
202

203
static void vnodeBufPoolResize(SVBufPool *pPool, int64_t size) {
×
204
  if (pPool == NULL) return;
×
205

206
  uint8_t *pDataNew = taosMemoryMalloc(size);
×
207
  if (pDataNew == NULL) {
×
208
    vError("vgId:%d, %s failed at %s:%d since %s", TD_VID(pPool->pVnode), __func__, __FILE__, __LINE__,
×
209
           tstrerror(terrno));
210
    return;
×
211
  }
212

213
  // Apply change
214
  int64_t oldSize = pPool->node.size;
×
215
  taosMemoryFree(pPool->node.data);
×
216
  pPool->node.data = pDataNew;
×
217
  pPool->node.size = size;
×
218

219
  vInfo("vgId:%d, buffer pool %d resized from %" PRId64 " to %" PRId64, TD_VID(pPool->pVnode), pPool->id, oldSize,
×
220
        size);
221
  return;
×
222
}
223

224
void vnodeBufPoolAddToFreeList(SVBufPool *pPool) {
5,429,224✔
225
  SVnode *pVnode = pPool->pVnode;
5,429,224✔
226

227
  int64_t size = pVnode->config.szBuf / VNODE_BUFPOOL_SEGMENTS;
5,429,224✔
228
  if (pPool->node.size != size) {
5,429,224✔
229
    vnodeBufPoolResize(pPool, size);
×
230
  }
231

232
  // add to free list
233
  vDebug("vgId:%d, buffer pool %p of id %d is added to free list", TD_VID(pVnode), pPool, pPool->id);
5,429,224✔
234
  vnodeBufPoolReset(pPool);
5,429,224✔
235
  pPool->freeNext = pVnode->freeList;
5,429,224✔
236
  pVnode->freeList = pPool;
5,429,224✔
237
  (void)taosThreadCondSignal(&pVnode->poolNotEmpty);
5,429,224✔
238
}
5,429,224✔
239

240
void vnodeBufPoolUnRef(SVBufPool *pPool, bool proactive) {
9,752,501✔
241
  if (pPool == NULL) return;
9,752,501✔
242

243
  SVnode *pVnode = pPool->pVnode;
9,752,501✔
244

245
  if (proactive) {
9,752,806✔
246
    (void)taosThreadMutexLock(&pVnode->mutex);
9,752,806✔
247
  }
248

249
  if (atomic_sub_fetch_32(&pPool->nRef, 1) > 0) goto _exit;
9,752,476✔
250

251
  // remove from recycle queue or on-recycle position
252
  if (pVnode->onRecycle == pPool) {
44,539✔
253
    pVnode->onRecycle = NULL;
×
254
  } else {
255
    if (pPool->recyclePrev) {
44,539✔
256
      pPool->recyclePrev->recycleNext = pPool->recycleNext;
×
257
    } else {
258
      pVnode->recycleHead = pPool->recycleNext;
44,539✔
259
    }
260

261
    if (pPool->recycleNext) {
44,539✔
UNCOV
262
      pPool->recycleNext->recyclePrev = pPool->recyclePrev;
×
263
    } else {
264
      pVnode->recycleTail = pPool->recyclePrev;
44,539✔
265
    }
266
    pPool->recyclePrev = pPool->recycleNext = NULL;
44,539✔
267
  }
268

269
  vnodeBufPoolAddToFreeList(pPool);
44,539✔
270

271
_exit:
9,752,806✔
272
  if (proactive) {
9,752,806✔
273
    (void)taosThreadMutexUnlock(&pVnode->mutex);
9,752,336✔
274
  }
275
  return;
9,752,110✔
276
}
277

278
void vnodeBufPoolRegisterQuery(SVBufPool *pPool, SQueryNode *pQNode) {
206,052,411✔
279
  (void)taosThreadMutexLock(&pPool->mutex);
206,052,411✔
280

281
  pQNode->pNext = pPool->qList.pNext;
206,084,852✔
282
  pQNode->ppNext = &pPool->qList.pNext;
206,079,404✔
283
  pPool->qList.pNext->ppNext = &pQNode->pNext;
206,072,028✔
284
  pPool->qList.pNext = pQNode;
206,041,996✔
285
  pPool->nQuery++;
206,058,416✔
286

287
  (void)taosThreadMutexUnlock(&pPool->mutex);
206,001,730✔
288
}
206,027,271✔
289

290
void vnodeBufPoolDeregisterQuery(SVBufPool *pPool, SQueryNode *pQNode, bool proactive) {
206,073,553✔
291
  int32_t code = 0;
206,073,553✔
292

293
  if (proactive) {
206,073,553✔
294
    (void)taosThreadMutexLock(&pPool->mutex);
206,081,286✔
295
  }
296

297
  pQNode->pNext->ppNext = pQNode->ppNext;
206,078,574✔
298
  *pQNode->ppNext = pQNode->pNext;
206,087,778✔
299
  pPool->nQuery--;
206,084,132✔
300

301
  if (proactive) {
206,078,266✔
302
    (void)taosThreadMutexUnlock(&pPool->mutex);
206,068,040✔
303
  }
304
}
206,098,293✔
305

306
int32_t vnodeBufPoolRecycle(SVBufPool *pPool) {
×
307
  int32_t code = 0;
×
308

309
  SVnode *pVnode = pPool->pVnode;
×
310

311
  vDebug("vgId:%d, recycle buffer pool %p of id %d", TD_VID(pVnode), pPool, pPool->id);
×
312

313
  (void)taosThreadMutexLock(&pPool->mutex);
×
314

315
  SQueryNode *pNode = pPool->qList.pNext;
×
316
  while (pNode != &pPool->qList) {
×
317
    SQueryNode *pTNode = pNode->pNext;
×
318

319
    int32_t rc = pNode->reseek(pNode->pQHandle);
×
320
    if (rc == 0 || rc == TSDB_CODE_VND_QUERY_BUSY) {
×
321
      pNode = pTNode;
×
322
    } else {
323
      code = rc;
×
324
      goto _exit;
×
325
    }
326
  }
327

328
_exit:
×
329
  (void)taosThreadMutexUnlock(&pPool->mutex);
×
330
  return code;
×
331
}
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