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

taosdata / TDengine / #4822

27 Oct 2025 05:42AM UTC coverage: 59.732% (+1.0%) from 58.728%
#4822

push

travis-ci

web-flow
Merge pull request #33377 from taosdata/fix/main/rename-udf-path

fix: update UDF example links to correct file paths

121214 of 258518 branches covered (46.89%)

Branch coverage included in aggregate %.

193636 of 268583 relevant lines covered (72.1%)

4002399.5 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) {
28,486✔
20
  SVBufPool *pPool;
21

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

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

48
  *ppPool = pPool;
28,480✔
49
  return 0;
28,480✔
50
}
51

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

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

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

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

80
void vnodeCloseBufPool(SVnode *pVnode) {
9,495✔
81
  for (int32_t i = 0; i < VNODE_BUFPOOL_SEGMENTS; i++) {
37,985✔
82
    if (pVnode->aBufPool[i]) {
28,490!
83
      vnodeBufPoolDestroy(pVnode->aBufPool[i]);
28,490✔
84
      pVnode->aBufPool[i] = NULL;
28,490✔
85
    }
86
  }
87

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

91
void vnodeBufPoolReset(SVBufPool *pPool) {
42,731✔
92
  if (pPool->nQuery != 0) {
42,731!
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,151,836✔
98
    pNode->prev->pnext = &pPool->pTail;
1,109,103✔
99
    pPool->pTail = pNode->prev;
1,109,103✔
100
    pPool->size = pPool->size - sizeof(*pNode) - pNode->size;
1,109,103✔
101
    taosMemoryFree(pNode);
1,109,103!
102
  }
103

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

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

114
  if (pPool == NULL) {
139,485,737!
115
    terrno = TSDB_CODE_INVALID_PARA;
×
116
    return NULL;
×
117
  }
118

119
  ptr = pPool->ptr;
139,485,737✔
120
  paddingLen = (((long)ptr + 7) & ~7) - (long)ptr;
139,485,737✔
121

122
  if (pPool->node.size >= pPool->ptr - pPool->node.data + size + paddingLen) {
139,485,737✔
123
    // allocate from the anchor node
124
    p = pPool->ptr + paddingLen;
138,389,239✔
125
    size += paddingLen;
138,389,239✔
126
    pPool->ptr = pPool->ptr + size;
138,389,239✔
127
    pPool->size += size;
138,389,239✔
128
  } else {
129
    // allocate a new node
130
    pNode = taosMemoryMalloc(sizeof(*pNode) + size);
1,096,498!
131
    if (pNode == NULL) {
1,109,103!
132
      return NULL;
×
133
    }
134
    pNode->data = (uint8_t *)&pNode[1];
1,109,103✔
135

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

143
    pPool->size = pPool->size + sizeof(*pNode) + size;
1,109,103✔
144
  }
145
  return p;
139,498,342✔
146
}
147

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

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

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

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

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

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

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

243
  SVnode *pVnode = pPool->pVnode;
23,739✔
244

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

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

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

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

269
  vnodeBufPoolAddToFreeList(pPool);
72✔
270

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

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

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

287
  (void)taosThreadMutexUnlock(&pPool->mutex);
294,464✔
288
}
294,449✔
289

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

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

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

301
  if (proactive) {
294,468!
302
    (void)taosThreadMutexUnlock(&pPool->mutex);
294,471✔
303
  }
304
}
294,463✔
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