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

taosdata / TDengine / #4118

17 May 2025 06:43AM UTC coverage: 62.797% (+0.7%) from 62.054%
#4118

push

travis-ci

web-flow
Merge pull request #31115 from taosdata/merge/mainto3.0

merge: from main to 3.0 branch

156841 of 318088 branches covered (49.31%)

Branch coverage included in aggregate %.

176 of 225 new or added lines in 20 files covered. (78.22%)

2989 existing lines in 163 files now uncovered.

242067 of 317143 relevant lines covered (76.33%)

6956088.31 hits per line

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

61.21
/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) {
42,337✔
20
  SVBufPool *pPool;
21

22
  pPool = taosMemoryMalloc(sizeof(SVBufPool) + size);
42,337!
23
  if (pPool == NULL) {
42,351!
24
    return terrno;
×
25
  }
26
  memset(pPool, 0, sizeof(SVBufPool));
42,351✔
27

28
  // query handle list
29
  (void)taosThreadMutexInit(&pPool->mutex, NULL);
42,351✔
30
  pPool->nQuery = 0;
42,341✔
31
  pPool->qList.pNext = &pPool->qList;
42,341✔
32
  pPool->qList.ppNext = &pPool->qList.pNext;
42,341✔
33

34
  pPool->pVnode = pVnode;
42,341✔
35
  pPool->id = id;
42,341✔
36
  pPool->ptr = pPool->node.data;
42,341✔
37
  pPool->pTail = &pPool->node;
42,341✔
38
  pPool->node.prev = NULL;
42,341✔
39
  pPool->node.pnext = &pPool->pTail;
42,341✔
40
  pPool->node.size = size;
42,341✔
41

42
  if (VND_IS_RSMA(pVnode)) {
42,341✔
43
    pPool->lock = taosMemoryMalloc(sizeof(TdThreadSpinlock));
60!
44
    if (!pPool->lock) {
63!
45
      taosMemoryFree(pPool);
×
46
      return terrno;
×
47
    }
48
    if (taosThreadSpinInit(pPool->lock, 0) != 0) {
63!
UNCOV
49
      taosMemoryFree((void *)pPool->lock);
×
50
      taosMemoryFree(pPool);
×
51
      return terrno = TAOS_SYSTEM_ERROR(ERRNO);
×
52
    }
53
  } else {
54
    pPool->lock = NULL;
42,281✔
55
  }
56

57
  *ppPool = pPool;
42,344✔
58
  return 0;
42,344✔
59
}
60

61
static void vnodeBufPoolDestroy(SVBufPool *pPool) {
42,356✔
62
  vnodeBufPoolReset(pPool);
42,356✔
63
  if (pPool->lock) {
42,357✔
64
    (void)taosThreadSpinDestroy(pPool->lock);
63✔
65
    taosMemoryFree((void *)pPool->lock);
63!
66
  }
67
  (void)taosThreadMutexDestroy(&pPool->mutex);
42,357✔
68
  taosMemoryFree(pPool);
42,357!
69
}
42,357✔
70

71
int vnodeOpenBufPool(SVnode *pVnode) {
14,119✔
72
  int64_t size = pVnode->config.szBuf / VNODE_BUFPOOL_SEGMENTS;
14,119✔
73

74
  for (int i = 0; i < VNODE_BUFPOOL_SEGMENTS; i++) {
56,463✔
75
    // create pool
76
    int32_t code;
77
    if ((code = vnodeBufPoolCreate(pVnode, i, size, &pVnode->aBufPool[i]))) {
42,338!
UNCOV
78
      vError("vgId:%d, failed to open vnode buffer pool since %s", TD_VID(pVnode), tstrerror(terrno));
×
UNCOV
79
      vnodeCloseBufPool(pVnode);
×
80
      return code;
×
81
    }
82

83
    // add to free list
84
    pVnode->aBufPool[i]->freeNext = pVnode->freeList;
42,344✔
85
    pVnode->freeList = pVnode->aBufPool[i];
42,344✔
86
  }
87

88
  vDebug("vgId:%d, vnode buffer pool is opened, size:%" PRId64, TD_VID(pVnode), size);
14,125✔
89
  return 0;
14,118✔
90
}
91

92
void vnodeCloseBufPool(SVnode *pVnode) {
14,118✔
93
  for (int32_t i = 0; i < VNODE_BUFPOOL_SEGMENTS; i++) {
56,475✔
94
    if (pVnode->aBufPool[i]) {
42,357!
95
      vnodeBufPoolDestroy(pVnode->aBufPool[i]);
42,357✔
96
      pVnode->aBufPool[i] = NULL;
42,357✔
97
    }
98
  }
99

100
  vDebug("vgId:%d, vnode buffer pool is closed", TD_VID(pVnode));
14,118✔
101
}
14,118✔
102

103
void vnodeBufPoolReset(SVBufPool *pPool) {
66,742✔
104
  if (pPool->nQuery != 0) {
66,742!
105
    vError("vgId:%d, buffer pool %p of id %d has %d queries, reset it may cause problem", TD_VID(pPool->pVnode), pPool,
×
106
           pPool->id, pPool->nQuery);
107
  }
108

109
  for (SVBufPoolNode *pNode = pPool->pTail; pNode->prev; pNode = pPool->pTail) {
10,037,590✔
110
    pNode->prev->pnext = &pPool->pTail;
9,970,846✔
111
    pPool->pTail = pNode->prev;
9,970,846✔
112
    pPool->size = pPool->size - sizeof(*pNode) - pNode->size;
9,970,846✔
113
    taosMemoryFree(pNode);
9,970,846!
114
  }
115

116
  pPool->size = 0;
66,744✔
117
  pPool->ptr = pPool->node.data;
66,744✔
118
}
66,744✔
119

120
void *vnodeBufPoolMallocAligned(SVBufPool *pPool, int size) {
207,391,203✔
121
  SVBufPoolNode *pNode;
122
  void          *p = NULL;
207,391,203✔
123
  uint8_t       *ptr = NULL;
207,391,203✔
124
  int            paddingLen = 0;
207,391,203✔
125

126
  if (pPool == NULL) {
207,391,203!
127
    terrno = TSDB_CODE_INVALID_PARA;
×
128
    return NULL;
×
129
  }
130

131
  if (pPool->lock) taosThreadSpinLock(pPool->lock);
207,391,203✔
132

133
  ptr = pPool->ptr;
207,319,306✔
134
  paddingLen = (((long)ptr + 7) & ~7) - (long)ptr;
207,319,306✔
135

136
  if (pPool->node.size >= pPool->ptr - pPool->node.data + size + paddingLen) {
207,319,306✔
137
    // allocate from the anchor node
138
    p = pPool->ptr + paddingLen;
197,348,460✔
139
    size += paddingLen;
197,348,460✔
140
    pPool->ptr = pPool->ptr + size;
197,348,460✔
141
    pPool->size += size;
197,348,460✔
142
  } else {
143
    // allocate a new node
144
    pNode = taosMemoryMalloc(sizeof(*pNode) + size);
9,970,846!
145
    if (pNode == NULL) {
9,970,846!
146
      if (pPool->lock) {
×
147
        (void)taosThreadSpinUnlock(pPool->lock);
×
148
      }
149
      return NULL;
×
150
    }
151

152
    p = pNode->data;
9,970,846✔
153
    pNode->size = size;
9,970,846✔
154
    pNode->prev = pPool->pTail;
9,970,846✔
155
    pNode->pnext = &pPool->pTail;
9,970,846✔
156
    pPool->pTail->pnext = &pNode->prev;
9,970,846✔
157
    pPool->pTail = pNode;
9,970,846✔
158

159
    pPool->size = pPool->size + sizeof(*pNode) + size;
9,970,846✔
160
  }
161
  if (pPool->lock) (void)taosThreadSpinUnlock(pPool->lock);
207,319,306✔
162
  return p;
207,980,868✔
163
}
164

165
void *vnodeBufPoolMalloc(SVBufPool *pPool, int size) {
92,748✔
166
  SVBufPoolNode *pNode;
167
  void          *p = NULL;
92,748✔
168

169
  if (pPool == NULL) {
92,748!
170
    terrno = TSDB_CODE_INVALID_PARA;
×
171
    return NULL;
×
172
  }
173

174
  if (pPool->lock) taosThreadSpinLock(pPool->lock);
92,748✔
175
  if (pPool->node.size >= pPool->ptr - pPool->node.data + size) {
92,742!
176
    // allocate from the anchor node
177
    p = pPool->ptr;
92,742✔
178
    pPool->ptr = pPool->ptr + size;
92,742✔
179
    pPool->size += size;
92,742✔
180
  } else {
181
    // allocate a new node
182
    pNode = taosMemoryMalloc(sizeof(*pNode) + size);
×
183
    if (pNode == NULL) {
×
184
      if (pPool->lock) (void)taosThreadSpinUnlock(pPool->lock);
×
185
      return NULL;
×
186
    }
187

188
    p = pNode->data;
×
189
    pNode->size = size;
×
190
    pNode->prev = pPool->pTail;
×
191
    pNode->pnext = &pPool->pTail;
×
192
    pPool->pTail->pnext = &pNode->prev;
×
193
    pPool->pTail = pNode;
×
194

195
    pPool->size = pPool->size + sizeof(*pNode) + size;
×
196
  }
197
  if (pPool->lock) (void)taosThreadSpinUnlock(pPool->lock);
92,742✔
198
  return p;
92,740✔
199
}
200

201
void vnodeBufPoolFree(SVBufPool *pPool, void *p) {
45,285✔
202
  // uint8_t       *ptr = (uint8_t *)p;
203
  // SVBufPoolNode *pNode;
204

205
  // if (ptr < pPool->node.data || ptr >= pPool->node.data + pPool->node.size) {
206
  //   pNode = &((SVBufPoolNode *)p)[-1];
207
  //   *pNode->pnext = pNode->prev;
208
  //   pNode->prev->pnext = pNode->pnext;
209

210
  //   pPool->size = pPool->size - sizeof(*pNode) - pNode->size;
211
  //   taosMemoryFree(pNode);
212
  // }
213
}
45,285✔
214

215
void vnodeBufPoolRef(SVBufPool *pPool) {
38,581✔
216
  int32_t nRef = atomic_fetch_add_32(&pPool->nRef, 1);
38,581✔
217
  if (nRef <= 0) {
38,582!
218
    vError("vgId:%d, buffer pool %p of id %d is referenced by %d", TD_VID(pPool->pVnode), pPool, pPool->id, nRef);
×
219
  }
220
}
38,582✔
221

222
void vnodeBufPoolAddToFreeList(SVBufPool *pPool) {
24,388✔
223
  SVnode *pVnode = pPool->pVnode;
24,388✔
224

225
  int64_t size = pVnode->config.szBuf / VNODE_BUFPOOL_SEGMENTS;
24,388✔
226
  if (pPool->node.size != size) {
24,388!
227
    SVBufPool *pNewPool = NULL;
×
228
    if (vnodeBufPoolCreate(pVnode, pPool->id, size, &pNewPool) < 0) {
×
229
      vWarn("vgId:%d, failed to change buffer pool of id %d size from %" PRId64 " to %" PRId64 " since %s",
×
230
            TD_VID(pVnode), pPool->id, pPool->node.size, size, tstrerror(ERRNO));
231
    } else {
232
      vInfo("vgId:%d, buffer pool of id %d size changed from %" PRId64 " to %" PRId64, TD_VID(pVnode), pPool->id,
×
233
            pPool->node.size, size);
234

235
      vnodeBufPoolDestroy(pPool);
×
236
      pPool = pNewPool;
×
237
      pVnode->aBufPool[pPool->id] = pPool;
×
238
    }
239
  }
240

241
  // add to free list
242
  vDebug("vgId:%d, buffer pool %p of id %d is added to free list", TD_VID(pVnode), pPool, pPool->id);
24,388✔
243
  vnodeBufPoolReset(pPool);
24,388✔
244
  pPool->freeNext = pVnode->freeList;
24,386✔
245
  pVnode->freeList = pPool;
24,386✔
246
  (void)taosThreadCondSignal(&pVnode->poolNotEmpty);
24,386✔
247
}
24,387✔
248

249
void vnodeBufPoolUnRef(SVBufPool *pPool, bool proactive) {
38,585✔
250
  if (pPool == NULL) return;
38,585!
251

252
  SVnode *pVnode = pPool->pVnode;
38,585✔
253

254
  if (proactive) {
38,585!
255
    (void)taosThreadMutexLock(&pVnode->mutex);
38,585✔
256
  }
257

258
  if (atomic_sub_fetch_32(&pPool->nRef, 1) > 0) goto _exit;
38,585✔
259

260
  // remove from recycle queue or on-recycle position
261
  if (pVnode->onRecycle == pPool) {
516!
262
    pVnode->onRecycle = NULL;
×
263
  } else {
264
    if (pPool->recyclePrev) {
516!
265
      pPool->recyclePrev->recycleNext = pPool->recycleNext;
×
266
    } else {
267
      pVnode->recycleHead = pPool->recycleNext;
516✔
268
    }
269

270
    if (pPool->recycleNext) {
516!
271
      pPool->recycleNext->recyclePrev = pPool->recyclePrev;
×
272
    } else {
273
      pVnode->recycleTail = pPool->recyclePrev;
516✔
274
    }
275
    pPool->recyclePrev = pPool->recycleNext = NULL;
516✔
276
  }
277

278
  vnodeBufPoolAddToFreeList(pPool);
516✔
279

280
_exit:
38,585✔
281
  if (proactive) {
38,585!
282
    (void)taosThreadMutexUnlock(&pVnode->mutex);
38,585✔
283
  }
284
  return;
38,584✔
285
}
286

287
void vnodeBufPoolRegisterQuery(SVBufPool *pPool, SQueryNode *pQNode) {
1,004,051✔
288
  (void)taosThreadMutexLock(&pPool->mutex);
1,004,051✔
289

290
  pQNode->pNext = pPool->qList.pNext;
1,004,628✔
291
  pQNode->ppNext = &pPool->qList.pNext;
1,004,628✔
292
  pPool->qList.pNext->ppNext = &pQNode->pNext;
1,004,628✔
293
  pPool->qList.pNext = pQNode;
1,004,628✔
294
  pPool->nQuery++;
1,004,628✔
295

296
  (void)taosThreadMutexUnlock(&pPool->mutex);
1,004,628✔
297
}
1,004,636✔
298

299
void vnodeBufPoolDeregisterQuery(SVBufPool *pPool, SQueryNode *pQNode, bool proactive) {
1,004,326✔
300
  int32_t code = 0;
1,004,326✔
301

302
  if (proactive) {
1,004,326!
303
    (void)taosThreadMutexLock(&pPool->mutex);
1,004,473✔
304
  }
305

306
  pQNode->pNext->ppNext = pQNode->ppNext;
1,004,523✔
307
  *pQNode->ppNext = pQNode->pNext;
1,004,523✔
308
  pPool->nQuery--;
1,004,523✔
309

310
  if (proactive) {
1,004,523!
311
    (void)taosThreadMutexUnlock(&pPool->mutex);
1,004,661✔
312
  }
313
}
1,004,529✔
314

315
int32_t vnodeBufPoolRecycle(SVBufPool *pPool) {
×
316
  int32_t code = 0;
×
317

318
  SVnode *pVnode = pPool->pVnode;
×
319

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

322
  (void)taosThreadMutexLock(&pPool->mutex);
×
323

324
  SQueryNode *pNode = pPool->qList.pNext;
×
325
  while (pNode != &pPool->qList) {
×
326
    SQueryNode *pTNode = pNode->pNext;
×
327

328
    int32_t rc = pNode->reseek(pNode->pQHandle);
×
329
    if (rc == 0 || rc == TSDB_CODE_VND_QUERY_BUSY) {
×
330
      pNode = pTNode;
×
331
    } else {
332
      code = rc;
×
333
      goto _exit;
×
334
    }
335
  }
336

337
_exit:
×
338
  (void)taosThreadMutexUnlock(&pPool->mutex);
×
339
  return code;
×
340
}
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