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

taosdata / TDengine / #4908

30 Dec 2025 10:52AM UTC coverage: 65.386% (-0.2%) from 65.541%
#4908

push

travis-ci

web-flow
enh: drop multi-stream (#33962)

60 of 106 new or added lines in 4 files covered. (56.6%)

1330 existing lines in 113 files now uncovered.

193461 of 295877 relevant lines covered (65.39%)

115765274.47 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) {
11,068,651✔
20
  SVBufPool *pPool;
21

22
  pPool = taosMemoryCalloc(1, sizeof(SVBufPool));
11,068,651✔
23
  if (pPool == NULL) {
11,069,049✔
24
    vError("vgId:%d, failed to allocate buffer pool", TD_VID(pVnode));
×
25
    return terrno;
×
26
  }
27
  pPool->node.data = taosMemoryMalloc(size);
11,069,049✔
28
  if (NULL == pPool->node.data) {
11,071,868✔
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);
11,071,667✔
36
  pPool->nQuery = 0;
11,070,044✔
37
  pPool->qList.pNext = &pPool->qList;
11,071,033✔
38
  pPool->qList.ppNext = &pPool->qList.pNext;
11,069,353✔
39

40
  pPool->pVnode = pVnode;
11,068,974✔
41
  pPool->id = id;
11,067,664✔
42
  pPool->ptr = pPool->node.data;
11,068,621✔
43
  pPool->pTail = &pPool->node;
11,070,749✔
44
  pPool->node.prev = NULL;
11,070,280✔
45
  pPool->node.pnext = &pPool->pTail;
11,068,114✔
46
  pPool->node.size = size;
11,069,238✔
47

48
  *ppPool = pPool;
11,068,341✔
49
  return 0;
11,068,630✔
50
}
51

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

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

62
  for (int i = 0; i < VNODE_BUFPOOL_SEGMENTS; i++) {
14,759,144✔
63
    // create pool
64
    int32_t code;
65
    if ((code = vnodeBufPoolCreate(pVnode, i, size, &pVnode->aBufPool[i]))) {
11,069,859✔
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;
11,068,630✔
73
    pVnode->freeList = pVnode->aBufPool[i];
11,067,753✔
74
  }
75

76
  vDebug("vgId:%d, vnode buffer pool is opened, size:%" PRId64, TD_VID(pVnode), size);
3,689,285✔
77
  return 0;
3,690,865✔
78
}
79

80
void vnodeCloseBufPool(SVnode *pVnode) {
3,690,865✔
81
  for (int32_t i = 0; i < VNODE_BUFPOOL_SEGMENTS; i++) {
14,763,460✔
82
    if (pVnode->aBufPool[i]) {
11,072,595✔
83
      vnodeBufPoolDestroy(pVnode->aBufPool[i]);
11,072,595✔
84
      pVnode->aBufPool[i] = NULL;
11,072,595✔
85
    }
86
  }
87

88
  vDebug("vgId:%d, vnode buffer pool is closed", TD_VID(pVnode));
3,690,865✔
89
}
3,690,865✔
90

91
void vnodeBufPoolReset(SVBufPool *pPool) {
15,927,449✔
92
  if (pPool->nQuery != 0) {
15,927,449✔
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) {
840,678,820✔
98
    pNode->prev->pnext = &pPool->pTail;
824,751,371✔
99
    pPool->pTail = pNode->prev;
824,751,371✔
100
    pPool->size = pPool->size - sizeof(*pNode) - pNode->size;
824,751,371✔
101
    taosMemoryFree(pNode);
824,751,371✔
102
  }
103

104
  pPool->size = 0;
15,927,449✔
105
  pPool->ptr = pPool->node.data;
15,927,449✔
106
}
15,927,449✔
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);
824,751,371✔
131
    if (pNode == NULL) {
824,751,371✔
132
      return NULL;
×
133
    }
134
    pNode->data = (uint8_t *)&pNode[1];
824,751,371✔
135

136
    p = pNode->data;
824,751,371✔
137
    pNode->size = size;
824,751,371✔
138
    pNode->prev = pPool->pTail;
824,751,371✔
139
    pNode->pnext = &pPool->pTail;
824,751,371✔
140
    pPool->pTail->pnext = &pNode->prev;
824,751,371✔
141
    pPool->pTail = pNode;
824,751,371✔
142

143
    pPool->size = pPool->size + sizeof(*pNode) + size;
824,751,371✔
144
  }
145
  return p;
2,147,483,647✔
146
}
147

148
void *vnodeBufPoolMalloc(SVBufPool *pPool, int size) {
21,538,481✔
149
  SVBufPoolNode *pNode;
150
  void          *p = NULL;
21,538,481✔
151

152
  if (pPool == NULL) {
21,538,481✔
153
    terrno = TSDB_CODE_INVALID_PARA;
×
154
    return NULL;
×
155
  }
156

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

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

UNCOV
177
    pPool->size = pPool->size + sizeof(*pNode) + size;
×
178
  }
179
  return p;
21,538,481✔
180
}
181

182
void vnodeBufPoolFree(SVBufPool *pPool, void *p) {
110,957,001✔
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
}
110,957,001✔
195

196
void vnodeBufPoolRef(SVBufPool *pPool) {
8,541,665✔
197
  int32_t nRef = atomic_fetch_add_32(&pPool->nRef, 1);
8,541,665✔
198
  if (nRef <= 0) {
8,542,023✔
199
    vError("vgId:%d, buffer pool %p of id %d is referenced by %d", TD_VID(pPool->pVnode), pPool, pPool->id, nRef);
×
200
  }
201
}
8,542,023✔
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) {
4,854,854✔
225
  SVnode *pVnode = pPool->pVnode;
4,854,854✔
226

227
  int64_t size = pVnode->config.szBuf / VNODE_BUFPOOL_SEGMENTS;
4,854,854✔
228
  if (pPool->node.size != size) {
4,854,854✔
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);
4,854,854✔
234
  vnodeBufPoolReset(pPool);
4,854,854✔
235
  pPool->freeNext = pVnode->freeList;
4,854,854✔
236
  pVnode->freeList = pPool;
4,854,854✔
237
  (void)taosThreadCondSignal(&pVnode->poolNotEmpty);
4,854,854✔
238
}
4,854,854✔
239

240
void vnodeBufPoolUnRef(SVBufPool *pPool, bool proactive) {
8,538,737✔
241
  if (pPool == NULL) return;
8,538,737✔
242

243
  SVnode *pVnode = pPool->pVnode;
8,538,737✔
244

245
  if (proactive) {
8,542,289✔
246
    (void)taosThreadMutexLock(&pVnode->mutex);
8,542,039✔
247
  }
248

249
  if (atomic_sub_fetch_32(&pPool->nRef, 1) > 0) goto _exit;
8,542,801✔
250

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

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

269
  vnodeBufPoolAddToFreeList(pPool);
30,073✔
270

271
_exit:
8,541,694✔
272
  if (proactive) {
8,541,694✔
273
    (void)taosThreadMutexUnlock(&pVnode->mutex);
8,542,289✔
274
  }
275
  return;
8,542,075✔
276
}
277

278
void vnodeBufPoolRegisterQuery(SVBufPool *pPool, SQueryNode *pQNode) {
183,807,898✔
279
  (void)taosThreadMutexLock(&pPool->mutex);
183,807,898✔
280

281
  pQNode->pNext = pPool->qList.pNext;
183,863,656✔
282
  pQNode->ppNext = &pPool->qList.pNext;
183,860,380✔
283
  pPool->qList.pNext->ppNext = &pQNode->pNext;
183,839,839✔
284
  pPool->qList.pNext = pQNode;
183,821,837✔
285
  pPool->nQuery++;
183,822,179✔
286

287
  (void)taosThreadMutexUnlock(&pPool->mutex);
183,801,166✔
288
}
183,793,705✔
289

290
void vnodeBufPoolDeregisterQuery(SVBufPool *pPool, SQueryNode *pQNode, bool proactive) {
183,796,281✔
291
  int32_t code = 0;
183,796,281✔
292

293
  if (proactive) {
183,796,281✔
294
    (void)taosThreadMutexLock(&pPool->mutex);
183,819,148✔
295
  }
296

297
  pQNode->pNext->ppNext = pQNode->ppNext;
183,848,070✔
298
  *pQNode->ppNext = pQNode->pNext;
183,863,356✔
299
  pPool->nQuery--;
183,850,116✔
300

301
  if (proactive) {
183,858,330✔
302
    (void)taosThreadMutexUnlock(&pPool->mutex);
183,864,276✔
303
  }
304
}
183,844,410✔
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