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

taosdata / TDengine / #4761

28 Sep 2025 10:49AM UTC coverage: 57.837% (-1.0%) from 58.866%
#4761

push

travis-ci

web-flow
merge: set version (#33122)

136913 of 302095 branches covered (45.32%)

Branch coverage included in aggregate %.

207750 of 293830 relevant lines covered (70.7%)

5673932.16 hits per line

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

57.69
/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) {
37,744✔
20
  SVBufPool *pPool;
21

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

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

48
  *ppPool = pPool;
37,743✔
49
  return 0;
37,743✔
50
}
51

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

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

62
  for (int i = 0; i < VNODE_BUFPOOL_SEGMENTS; i++) {
50,332✔
63
    // create pool
64
    int32_t code;
65
    if ((code = vnodeBufPoolCreate(pVnode, i, size, &pVnode->aBufPool[i]))) {
37,744!
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;
37,747✔
73
    pVnode->freeList = pVnode->aBufPool[i];
37,747✔
74
  }
75

76
  vDebug("vgId:%d, vnode buffer pool is opened, size:%" PRId64, TD_VID(pVnode), size);
12,588✔
77
  return 0;
12,585✔
78
}
79

80
void vnodeCloseBufPool(SVnode *pVnode) {
12,584✔
81
  for (int32_t i = 0; i < VNODE_BUFPOOL_SEGMENTS; i++) {
50,339✔
82
    if (pVnode->aBufPool[i]) {
37,754!
83
      vnodeBufPoolDestroy(pVnode->aBufPool[i]);
37,754✔
84
      pVnode->aBufPool[i] = NULL;
37,755✔
85
    }
86
  }
87

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

91
void vnodeBufPoolReset(SVBufPool *pPool) {
59,599✔
92
  if (pPool->nQuery != 0) {
59,599!
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) {
262,543✔
98
    pNode->prev->pnext = &pPool->pTail;
202,943✔
99
    pPool->pTail = pNode->prev;
202,943✔
100
    pPool->size = pPool->size - sizeof(*pNode) - pNode->size;
202,943✔
101
    taosMemoryFree(pNode);
202,943!
102
  }
103

104
  pPool->size = 0;
59,600✔
105
  pPool->ptr = pPool->node.data;
59,600✔
106
}
59,600✔
107

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

114
  if (pPool == NULL) {
121,853,053!
115
    terrno = TSDB_CODE_INVALID_PARA;
×
116
    return NULL;
×
117
  }
118

119
  ptr = pPool->ptr;
121,853,053✔
120
  paddingLen = (((long)ptr + 7) & ~7) - (long)ptr;
121,853,053✔
121

122
  if (pPool->node.size >= pPool->ptr - pPool->node.data + size + paddingLen) {
121,853,053✔
123
    // allocate from the anchor node
124
    p = pPool->ptr + paddingLen;
121,671,249✔
125
    size += paddingLen;
121,671,249✔
126
    pPool->ptr = pPool->ptr + size;
121,671,249✔
127
    pPool->size += size;
121,671,249✔
128
  } else {
129
    // allocate a new node
130
    pNode = taosMemoryMalloc(sizeof(*pNode) + size);
181,804!
131
    if (pNode == NULL) {
202,943!
132
      return NULL;
×
133
    }
134
    pNode->data = (uint8_t *)&pNode[1];
202,943✔
135

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

143
    pPool->size = pPool->size + sizeof(*pNode) + size;
202,943✔
144
  }
145
  return p;
121,874,192✔
146
}
147

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

152
  if (pPool == NULL) {
19,635!
153
    terrno = TSDB_CODE_INVALID_PARA;
×
154
    return NULL;
×
155
  }
156

157
  if (pPool->node.size >= pPool->ptr - pPool->node.data + size) {
19,635!
158
    // allocate from the anchor node
159
    p = pPool->ptr;
19,636✔
160
    pPool->ptr = pPool->ptr + size;
19,636✔
161
    pPool->size += size;
19,636✔
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;
19,636✔
180
}
181

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

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

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

240
void vnodeBufPoolUnRef(SVBufPool *pPool, bool proactive) {
34,427✔
241
  if (pPool == NULL) return;
34,427!
242

243
  SVnode *pVnode = pPool->pVnode;
34,427✔
244

245
  if (proactive) {
34,427!
246
    (void)taosThreadMutexLock(&pVnode->mutex);
34,429✔
247
  }
248

249
  if (atomic_sub_fetch_32(&pPool->nRef, 1) > 0) goto _exit;
34,428✔
250

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

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

269
  vnodeBufPoolAddToFreeList(pPool);
97✔
270

271
_exit:
34,430✔
272
  if (proactive) {
34,430!
273
    (void)taosThreadMutexUnlock(&pVnode->mutex);
34,430✔
274
  }
275
  return;
34,430✔
276
}
277

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

281
  pQNode->pNext = pPool->qList.pNext;
858,031✔
282
  pQNode->ppNext = &pPool->qList.pNext;
858,031✔
283
  pPool->qList.pNext->ppNext = &pQNode->pNext;
858,031✔
284
  pPool->qList.pNext = pQNode;
858,031✔
285
  pPool->nQuery++;
858,031✔
286

287
  (void)taosThreadMutexUnlock(&pPool->mutex);
858,031✔
288
}
858,009✔
289

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

293
  if (proactive) {
858,020!
294
    (void)taosThreadMutexLock(&pPool->mutex);
858,024✔
295
  }
296

297
  pQNode->pNext->ppNext = pQNode->ppNext;
858,029✔
298
  *pQNode->ppNext = pQNode->pNext;
858,029✔
299
  pPool->nQuery--;
858,029✔
300

301
  if (proactive) {
858,029!
302
    (void)taosThreadMutexUnlock(&pPool->mutex);
858,032✔
303
  }
304
}
858,028✔
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