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

taosdata / TDengine / #4679

18 Aug 2025 07:58AM UTC coverage: 60.031% (+0.02%) from 60.007%
#4679

push

travis-ci

web-flow
test: update case desc (#32551)

137751 of 292075 branches covered (47.16%)

Branch coverage included in aggregate %.

208308 of 284395 relevant lines covered (73.25%)

20959061.42 hits per line

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

51.31
/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) {
41,000✔
20
  SVBufPool *pPool;
21

22
  pPool = taosMemoryCalloc(1, sizeof(SVBufPool));
41,000!
23
  if (pPool == NULL) {
41,001!
24
    vError("vgId:%d, failed to allocate buffer pool", TD_VID(pVnode));
×
25
    return terrno;
×
26
  }
27
  pPool->node.data = taosMemoryMalloc(size);
41,001!
28
  if (NULL == pPool->node.data) {
41,000!
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);
41,000✔
36
  pPool->nQuery = 0;
40,999✔
37
  pPool->qList.pNext = &pPool->qList;
40,999✔
38
  pPool->qList.ppNext = &pPool->qList.pNext;
40,999✔
39

40
  pPool->pVnode = pVnode;
40,999✔
41
  pPool->id = id;
40,999✔
42
  pPool->ptr = pPool->node.data;
40,999✔
43
  pPool->pTail = &pPool->node;
40,999✔
44
  pPool->node.prev = NULL;
40,999✔
45
  pPool->node.pnext = &pPool->pTail;
40,999✔
46
  pPool->node.size = size;
40,999✔
47

48
  if (VND_IS_RSMA(pVnode)) {
40,999!
49
    pPool->lock = taosMemoryMalloc(sizeof(TdThreadSpinlock));
×
50
    if (!pPool->lock) {
×
51
      taosMemoryFree(pPool->node.data);
×
52
      taosMemoryFree(pPool);
×
53
      return terrno;
×
54
    }
55
    if (taosThreadSpinInit(pPool->lock, 0) != 0) {
×
56
      taosMemoryFree((void *)pPool->lock);
6!
57
      taosMemoryFree(pPool->node.data);
×
58
      taosMemoryFree(pPool);
×
59
      return terrno = TAOS_SYSTEM_ERROR(ERRNO);
×
60
    }
61
  } else {
62
    pPool->lock = NULL;
40,999✔
63
  }
64

65
  *ppPool = pPool;
40,993✔
66
  return 0;
40,993✔
67
}
68

69
static void vnodeBufPoolDestroy(SVBufPool *pPool) {
41,013✔
70
  vnodeBufPoolReset(pPool);
41,013✔
71
  if (pPool->lock) {
41,012!
72
    (void)taosThreadSpinDestroy(pPool->lock);
×
73
    taosMemoryFree((void *)pPool->lock);
×
74
  }
75
  (void)taosThreadMutexDestroy(&pPool->mutex);
41,012✔
76
  taosMemoryFree(pPool->node.data);
41,015!
77
  taosMemoryFree(pPool);
41,016!
78
}
41,016✔
79

80
int vnodeOpenBufPool(SVnode *pVnode) {
13,672✔
81
  int64_t size = pVnode->config.szBuf / VNODE_BUFPOOL_SEGMENTS;
13,672✔
82

83
  for (int i = 0; i < VNODE_BUFPOOL_SEGMENTS; i++) {
54,672✔
84
    // create pool
85
    int32_t code;
86
    if ((code = vnodeBufPoolCreate(pVnode, i, size, &pVnode->aBufPool[i]))) {
41,000!
87
      vError("vgId:%d, failed to open vnode buffer pool since %s", TD_VID(pVnode), tstrerror(terrno));
×
88
      vnodeCloseBufPool(pVnode);
×
89
      return code;
×
90
    }
91

92
    // add to free list
93
    pVnode->aBufPool[i]->freeNext = pVnode->freeList;
41,000✔
94
    pVnode->freeList = pVnode->aBufPool[i];
41,000✔
95
  }
96

97
  vDebug("vgId:%d, vnode buffer pool is opened, size:%" PRId64, TD_VID(pVnode), size);
13,672✔
98
  return 0;
13,672✔
99
}
100

101
void vnodeCloseBufPool(SVnode *pVnode) {
13,671✔
102
  for (int32_t i = 0; i < VNODE_BUFPOOL_SEGMENTS; i++) {
54,687✔
103
    if (pVnode->aBufPool[i]) {
41,014!
104
      vnodeBufPoolDestroy(pVnode->aBufPool[i]);
41,014✔
105
      pVnode->aBufPool[i] = NULL;
41,016✔
106
    }
107
  }
108

109
  vDebug("vgId:%d, vnode buffer pool is closed", TD_VID(pVnode));
13,673✔
110
}
13,673✔
111

112
void vnodeBufPoolReset(SVBufPool *pPool) {
73,814✔
113
  if (pPool->nQuery != 0) {
73,814!
114
    vError("vgId:%d, buffer pool %p of id %d has %d queries, reset it may cause problem", TD_VID(pPool->pVnode), pPool,
×
115
           pPool->id, pPool->nQuery);
116
  }
117

118
  for (SVBufPoolNode *pNode = pPool->pTail; pNode->prev; pNode = pPool->pTail) {
2,208,941✔
119
    pNode->prev->pnext = &pPool->pTail;
2,135,125✔
120
    pPool->pTail = pNode->prev;
2,135,125✔
121
    pPool->size = pPool->size - sizeof(*pNode) - pNode->size;
2,135,125✔
122
    taosMemoryFree(pNode);
2,135,125!
123
  }
124

125
  pPool->size = 0;
73,816✔
126
  pPool->ptr = pPool->node.data;
73,816✔
127
}
73,816✔
128

129
void *vnodeBufPoolMallocAligned(SVBufPool *pPool, int size) {
657,580,005✔
130
  SVBufPoolNode *pNode;
131
  void          *p = NULL;
657,580,005✔
132
  uint8_t       *ptr = NULL;
657,580,005✔
133
  int            paddingLen = 0;
657,580,005✔
134

135
  if (pPool == NULL) {
657,580,005!
136
    terrno = TSDB_CODE_INVALID_PARA;
×
137
    return NULL;
×
138
  }
139

140
  if (pPool->lock) taosThreadSpinLock(pPool->lock);
657,580,005!
141

142
  ptr = pPool->ptr;
657,631,658✔
143
  paddingLen = (((long)ptr + 7) & ~7) - (long)ptr;
657,631,658✔
144

145
  if (pPool->node.size >= pPool->ptr - pPool->node.data + size + paddingLen) {
657,631,658✔
146
    // allocate from the anchor node
147
    p = pPool->ptr + paddingLen;
655,496,533✔
148
    size += paddingLen;
655,496,533✔
149
    pPool->ptr = pPool->ptr + size;
655,496,533✔
150
    pPool->size += size;
655,496,533✔
151
  } else {
152
    // allocate a new node
153
    pNode = taosMemoryMalloc(sizeof(*pNode) + size);
2,135,125!
154
    if (pNode == NULL) {
2,135,125!
155
      if (pPool->lock) {
×
156
        (void)taosThreadSpinUnlock(pPool->lock);
×
157
      }
158
      return NULL;
×
159
    }
160
    pNode->data = (uint8_t *)&pNode[1];
2,135,125✔
161

162
    p = pNode->data;
2,135,125✔
163
    pNode->size = size;
2,135,125✔
164
    pNode->prev = pPool->pTail;
2,135,125✔
165
    pNode->pnext = &pPool->pTail;
2,135,125✔
166
    pPool->pTail->pnext = &pNode->prev;
2,135,125✔
167
    pPool->pTail = pNode;
2,135,125✔
168

169
    pPool->size = pPool->size + sizeof(*pNode) + size;
2,135,125✔
170
  }
171
  if (pPool->lock) (void)taosThreadSpinUnlock(pPool->lock);
657,631,658!
172
  return p;
657,600,760✔
173
}
174

175
void *vnodeBufPoolMalloc(SVBufPool *pPool, int size) {
133,751✔
176
  SVBufPoolNode *pNode;
177
  void          *p = NULL;
133,751✔
178

179
  if (pPool == NULL) {
133,751!
180
    terrno = TSDB_CODE_INVALID_PARA;
×
181
    return NULL;
×
182
  }
183

184
  if (pPool->lock) taosThreadSpinLock(pPool->lock);
133,751!
185
  if (pPool->node.size >= pPool->ptr - pPool->node.data + size) {
133,751!
186
    // allocate from the anchor node
187
    p = pPool->ptr;
133,751✔
188
    pPool->ptr = pPool->ptr + size;
133,751✔
189
    pPool->size += size;
133,751✔
190
  } else {
191
    // allocate a new node
192
    pNode = taosMemoryMalloc(sizeof(*pNode) + size);
×
193
    if (pNode == NULL) {
×
194
      if (pPool->lock) (void)taosThreadSpinUnlock(pPool->lock);
×
195
      return NULL;
×
196
    }
197
    pNode->data = (uint8_t *)&pNode[1];
×
198

199
    p = pNode->data;
×
200
    pNode->size = size;
×
201
    pNode->prev = pPool->pTail;
×
202
    pNode->pnext = &pPool->pTail;
×
203
    pPool->pTail->pnext = &pNode->prev;
×
204
    pPool->pTail = pNode;
×
205

206
    pPool->size = pPool->size + sizeof(*pNode) + size;
×
207
  }
208
  if (pPool->lock) (void)taosThreadSpinUnlock(pPool->lock);
133,751!
209
  return p;
133,750✔
210
}
211

212
void vnodeBufPoolFree(SVBufPool *pPool, void *p) {
179,306✔
213
  // uint8_t       *ptr = (uint8_t *)p;
214
  // SVBufPoolNode *pNode;
215

216
  // if (ptr < pPool->node.data || ptr >= pPool->node.data + pPool->node.size) {
217
  //   pNode = &((SVBufPoolNode *)p)[-1];
218
  //   *pNode->pnext = pNode->prev;
219
  //   pNode->prev->pnext = pNode->pnext;
220

221
  //   pPool->size = pPool->size - sizeof(*pNode) - pNode->size;
222
  //   taosMemoryFree(pNode);
223
  // }
224
}
179,306✔
225

226
void vnodeBufPoolRef(SVBufPool *pPool) {
46,472✔
227
  int32_t nRef = atomic_fetch_add_32(&pPool->nRef, 1);
46,472✔
228
  if (nRef <= 0) {
46,474!
229
    vError("vgId:%d, buffer pool %p of id %d is referenced by %d", TD_VID(pPool->pVnode), pPool, pPool->id, nRef);
×
230
  }
231
}
46,474✔
232

233
static void vnodeBufPoolResize(SVBufPool *pPool, int64_t size) {
×
234
  if (pPool == NULL) return;
×
235

236
  uint8_t *pDataNew = taosMemoryMalloc(size);
×
237
  if (pDataNew == NULL) {
×
238
    vError("vgId:%d, %s failed at %s:%d since %s", TD_VID(pPool->pVnode), __func__, __FILE__, __LINE__,
×
239
           tstrerror(terrno));
240
    return;
×
241
  }
242

243
  // Apply change
244
  int64_t oldSize = pPool->node.size;
×
245
  taosMemoryFree(pPool->node.data);
×
246
  pPool->node.data = pDataNew;
×
247
  pPool->node.size = size;
×
248

249
  vInfo("vgId:%d, buffer pool %d resized from %" PRId64 " to %" PRId64, TD_VID(pPool->pVnode), pPool->id, oldSize,
×
250
        size);
251
  return;
×
252
}
253

254
void vnodeBufPoolAddToFreeList(SVBufPool *pPool) {
32,802✔
255
  SVnode *pVnode = pPool->pVnode;
32,802✔
256

257
  int64_t size = pVnode->config.szBuf / VNODE_BUFPOOL_SEGMENTS;
32,802✔
258
  if (pPool->node.size != size) {
32,802!
259
    vnodeBufPoolResize(pPool, size);
×
260
  }
261

262
  // add to free list
263
  vDebug("vgId:%d, buffer pool %p of id %d is added to free list", TD_VID(pVnode), pPool, pPool->id);
32,802✔
264
  vnodeBufPoolReset(pPool);
32,802✔
265
  pPool->freeNext = pVnode->freeList;
32,800✔
266
  pVnode->freeList = pPool;
32,800✔
267
  (void)taosThreadCondSignal(&pVnode->poolNotEmpty);
32,800✔
268
}
32,802✔
269

270
void vnodeBufPoolUnRef(SVBufPool *pPool, bool proactive) {
46,473✔
271
  if (pPool == NULL) return;
46,473!
272

273
  SVnode *pVnode = pPool->pVnode;
46,473✔
274

275
  if (proactive) {
46,473!
276
    (void)taosThreadMutexLock(&pVnode->mutex);
46,473✔
277
  }
278

279
  if (atomic_sub_fetch_32(&pPool->nRef, 1) > 0) goto _exit;
46,474✔
280

281
  // remove from recycle queue or on-recycle position
282
  if (pVnode->onRecycle == pPool) {
348!
283
    pVnode->onRecycle = NULL;
×
284
  } else {
285
    if (pPool->recyclePrev) {
348!
286
      pPool->recyclePrev->recycleNext = pPool->recycleNext;
×
287
    } else {
288
      pVnode->recycleHead = pPool->recycleNext;
348✔
289
    }
290

291
    if (pPool->recycleNext) {
348!
292
      pPool->recycleNext->recyclePrev = pPool->recyclePrev;
×
293
    } else {
294
      pVnode->recycleTail = pPool->recyclePrev;
348✔
295
    }
296
    pPool->recyclePrev = pPool->recycleNext = NULL;
348✔
297
  }
298

299
  vnodeBufPoolAddToFreeList(pPool);
348✔
300

301
_exit:
46,474✔
302
  if (proactive) {
46,474!
303
    (void)taosThreadMutexUnlock(&pVnode->mutex);
46,474✔
304
  }
305
  return;
46,474✔
306
}
307

308
void vnodeBufPoolRegisterQuery(SVBufPool *pPool, SQueryNode *pQNode) {
6,617,218✔
309
  (void)taosThreadMutexLock(&pPool->mutex);
6,617,218✔
310

311
  pQNode->pNext = pPool->qList.pNext;
6,626,500✔
312
  pQNode->ppNext = &pPool->qList.pNext;
6,626,500✔
313
  pPool->qList.pNext->ppNext = &pQNode->pNext;
6,626,500✔
314
  pPool->qList.pNext = pQNode;
6,626,500✔
315
  pPool->nQuery++;
6,626,500✔
316

317
  (void)taosThreadMutexUnlock(&pPool->mutex);
6,626,500✔
318
}
6,625,072✔
319

320
void vnodeBufPoolDeregisterQuery(SVBufPool *pPool, SQueryNode *pQNode, bool proactive) {
6,626,588✔
321
  int32_t code = 0;
6,626,588✔
322

323
  if (proactive) {
6,626,588!
324
    (void)taosThreadMutexLock(&pPool->mutex);
6,626,715✔
325
  }
326

327
  pQNode->pNext->ppNext = pQNode->ppNext;
6,626,855✔
328
  *pQNode->ppNext = pQNode->pNext;
6,626,855✔
329
  pPool->nQuery--;
6,626,855✔
330

331
  if (proactive) {
6,626,855!
332
    (void)taosThreadMutexUnlock(&pPool->mutex);
6,627,011✔
333
  }
334
}
6,626,808✔
335

336
int32_t vnodeBufPoolRecycle(SVBufPool *pPool) {
×
337
  int32_t code = 0;
×
338

339
  SVnode *pVnode = pPool->pVnode;
×
340

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

343
  (void)taosThreadMutexLock(&pPool->mutex);
×
344

345
  SQueryNode *pNode = pPool->qList.pNext;
×
346
  while (pNode != &pPool->qList) {
×
347
    SQueryNode *pTNode = pNode->pNext;
×
348

349
    int32_t rc = pNode->reseek(pNode->pQHandle);
×
350
    if (rc == 0 || rc == TSDB_CODE_VND_QUERY_BUSY) {
×
351
      pNode = pTNode;
×
352
    } else {
353
      code = rc;
×
354
      goto _exit;
×
355
    }
356
  }
357

358
_exit:
×
359
  (void)taosThreadMutexUnlock(&pPool->mutex);
×
360
  return code;
×
361
}
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