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

taosdata / TDengine / #3593

24 Jan 2025 08:57AM UTC coverage: 63.239% (-0.3%) from 63.546%
#3593

push

travis-ci

web-flow
Merge pull request #29638 from taosdata/docs/TS-5846-3.0

enh: TDengine modify taosBenchmark new query rule cases and add doc

140619 of 285630 branches covered (49.23%)

Branch coverage included in aggregate %.

218877 of 282844 relevant lines covered (77.38%)

19647377.39 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) {
36,587✔
20
  SVBufPool *pPool;
21

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

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

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

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

57
  *ppPool = pPool;
36,589✔
58
  return 0;
36,589✔
59
}
60

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

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

74
  for (int i = 0; i < VNODE_BUFPOOL_SEGMENTS; i++) {
48,787✔
75
    // create pool
76
    int32_t code;
77
    if ((code = vnodeBufPoolCreate(pVnode, i, size, &pVnode->aBufPool[i]))) {
36,587!
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;
36,589✔
85
    pVnode->freeList = pVnode->aBufPool[i];
36,589✔
86
  }
87

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

92
void vnodeCloseBufPool(SVnode *pVnode) {
12,198✔
93
  for (int32_t i = 0; i < VNODE_BUFPOOL_SEGMENTS; i++) {
48,792✔
94
    if (pVnode->aBufPool[i]) {
36,594!
95
      vnodeBufPoolDestroy(pVnode->aBufPool[i]);
36,594✔
96
      pVnode->aBufPool[i] = NULL;
36,594✔
97
    }
98
  }
99

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

103
void vnodeBufPoolReset(SVBufPool *pPool) {
66,436✔
104
  if (pPool->nQuery != 0) {
66,436!
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) {
1,052,635✔
110
    pNode->prev->pnext = &pPool->pTail;
986,199✔
111
    pPool->pTail = pNode->prev;
986,199✔
112
    pPool->size = pPool->size - sizeof(*pNode) - pNode->size;
986,199✔
113
    taosMemoryFree(pNode);
986,199!
114
  }
115

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

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

126
  if (pPool == NULL) {
801,291,036!
127
    terrno = TSDB_CODE_INVALID_PARA;
×
128
    return NULL;
×
129
  }
130

131
  if (pPool->lock) taosThreadSpinLock(pPool->lock);
801,291,036✔
132

133
  ptr = pPool->ptr;
801,309,556✔
134
  paddingLen = (((long)ptr + 7) & ~7) - (long)ptr;
801,309,556✔
135

136
  if (pPool->node.size >= pPool->ptr - pPool->node.data + size + paddingLen) {
801,309,556✔
137
    // allocate from the anchor node
138
    p = pPool->ptr + paddingLen;
800,323,357✔
139
    size += paddingLen;
800,323,357✔
140
    pPool->ptr = pPool->ptr + size;
800,323,357✔
141
    pPool->size += size;
800,323,357✔
142
  } else {
143
    // allocate a new node
144
    pNode = taosMemoryMalloc(sizeof(*pNode) + size);
986,199!
145
    if (pNode == NULL) {
986,199!
146
      if (pPool->lock) {
×
147
        (void)taosThreadSpinUnlock(pPool->lock);
×
148
      }
149
      return NULL;
×
150
    }
151

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

159
    pPool->size = pPool->size + sizeof(*pNode) + size;
986,199✔
160
  }
161
  if (pPool->lock) (void)taosThreadSpinUnlock(pPool->lock);
801,309,556✔
162
  return p;
801,321,087✔
163
}
164

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

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

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

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

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

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

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

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

252
  SVnode *pVnode = pPool->pVnode;
42,102✔
253

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

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

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

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

278
  vnodeBufPoolAddToFreeList(pPool);
664✔
279

280
_exit:
42,103✔
281
  if (proactive) {
42,103✔
282
    (void)taosThreadMutexUnlock(&pVnode->mutex);
42,102✔
283
  }
284
  return;
42,103✔
285
}
286

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

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

296
  (void)taosThreadMutexUnlock(&pPool->mutex);
3,973,048✔
297
}
3,972,932✔
298

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

302
  if (proactive) {
3,972,749!
303
    (void)taosThreadMutexLock(&pPool->mutex);
3,973,040✔
304
  }
305

306
  pQNode->pNext->ppNext = pQNode->ppNext;
3,973,169✔
307
  *pQNode->ppNext = pQNode->pNext;
3,973,169✔
308
  pPool->nQuery--;
3,973,169✔
309

310
  if (proactive) {
3,973,169!
311
    (void)taosThreadMutexUnlock(&pPool->mutex);
3,973,473✔
312
  }
313
}
3,973,155✔
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