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

taosdata / TDengine / #3608

12 Feb 2025 05:57AM UTC coverage: 63.066% (+1.4%) from 61.715%
#3608

push

travis-ci

web-flow
Merge pull request #29746 from taosdata/merge/mainto3.02

merge: from main to 3.0 branch

140199 of 286257 branches covered (48.98%)

Branch coverage included in aggregate %.

89 of 161 new or added lines in 18 files covered. (55.28%)

3211 existing lines in 190 files now uncovered.

218998 of 283298 relevant lines covered (77.3%)

5949310.66 hits per line

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

61.52
/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) {
32,321✔
20
  SVBufPool *pPool;
21

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

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

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

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

57
  *ppPool = pPool;
32,321✔
58
  return 0;
32,321✔
59
}
60

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

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

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

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

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

92
void vnodeCloseBufPool(SVnode *pVnode) {
10,774✔
93
  for (int32_t i = 0; i < VNODE_BUFPOOL_SEGMENTS; i++) {
43,099✔
94
    if (pVnode->aBufPool[i]) {
32,325!
95
      vnodeBufPoolDestroy(pVnode->aBufPool[i]);
32,325✔
96
      pVnode->aBufPool[i] = NULL;
32,325✔
97
    }
98
  }
99

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

103
void vnodeBufPoolReset(SVBufPool *pPool) {
55,076✔
104
  if (pPool->nQuery != 0) {
55,076!
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) {
218,048✔
110
    pNode->prev->pnext = &pPool->pTail;
162,970✔
111
    pPool->pTail = pNode->prev;
162,970✔
112
    pPool->size = pPool->size - sizeof(*pNode) - pNode->size;
162,970✔
113
    taosMemoryFree(pNode);
162,970!
114
  }
115

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

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

126
  if (pPool == NULL) {
111,392,455!
127
    terrno = TSDB_CODE_INVALID_PARA;
×
128
    return NULL;
×
129
  }
130

131
  if (pPool->lock) taosThreadSpinLock(pPool->lock);
111,392,455✔
132

133
  ptr = pPool->ptr;
111,407,528✔
134
  paddingLen = (((long)ptr + 7) & ~7) - (long)ptr;
111,407,528✔
135

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

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

159
    pPool->size = pPool->size + sizeof(*pNode) + size;
162,970✔
160
  }
161
  if (pPool->lock) (void)taosThreadSpinUnlock(pPool->lock);
111,407,528✔
162
  return p;
111,405,566✔
163
}
164

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

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

174
  if (pPool->lock) taosThreadSpinLock(pPool->lock);
8,859✔
175
  if (pPool->node.size >= pPool->ptr - pPool->node.data + size) {
8,859!
176
    // allocate from the anchor node
177
    p = pPool->ptr;
8,859✔
178
    pPool->ptr = pPool->ptr + size;
8,859✔
179
    pPool->size += size;
8,859✔
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);
8,859✔
198
  return p;
8,859✔
199
}
200

201
void vnodeBufPoolFree(SVBufPool *pPool, void *p) {
42,668✔
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
}
42,668✔
214

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

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

225
  int64_t size = pVnode->config.szBuf / VNODE_BUFPOOL_SEGMENTS;
22,755✔
226
  if (pPool->node.size != size) {
22,755!
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);
22,755✔
243
  vnodeBufPoolReset(pPool);
22,755✔
244
  pPool->freeNext = pVnode->freeList;
22,753✔
245
  pVnode->freeList = pPool;
22,753✔
246
  (void)taosThreadCondSignal(&pVnode->poolNotEmpty);
22,753✔
247
}
22,753✔
248

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

252
  SVnode *pVnode = pPool->pVnode;
33,591✔
253

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

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

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

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

278
  vnodeBufPoolAddToFreeList(pPool);
560✔
279

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

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

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

296
  (void)taosThreadMutexUnlock(&pPool->mutex);
796,351✔
297
}
796,316✔
298

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

302
  if (proactive) {
796,194!
303
    (void)taosThreadMutexLock(&pPool->mutex);
796,243✔
304
  }
305

306
  pQNode->pNext->ppNext = pQNode->ppNext;
796,297✔
307
  *pQNode->ppNext = pQNode->pNext;
796,297✔
308
  pPool->nQuery--;
796,297✔
309

310
  if (proactive) {
796,297!
311
    (void)taosThreadMutexUnlock(&pPool->mutex);
796,345✔
312
  }
313
}
796,312✔
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