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

taosdata / TDengine / #3523

06 Nov 2024 02:29AM UTC coverage: 55.861% (-2.4%) from 58.216%
#3523

push

travis-ci

web-flow
Merge pull request #28551 from taosdata/feat/TS-5215-2

test(blob): testing & fixes for blob

106075 of 245834 branches covered (43.15%)

Branch coverage included in aggregate %.

0 of 15 new or added lines in 2 files covered. (0.0%)

17003 existing lines in 254 files now uncovered.

181910 of 269703 relevant lines covered (67.45%)

1527639.59 hits per line

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

64.83
/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) {
16,187✔
20
  SVBufPool *pPool;
21

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

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

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

42
  if (VND_IS_RSMA(pVnode)) {
16,204✔
43
    pPool->lock = taosMemoryMalloc(sizeof(TdThreadSpinlock));
53✔
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;
16,151✔
55
  }
56

57
  *ppPool = pPool;
16,208✔
58
  return 0;
16,208✔
59
}
60

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

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

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

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

92
void vnodeCloseBufPool(SVnode *pVnode) {
5,406✔
93
  for (int32_t i = 0; i < VNODE_BUFPOOL_SEGMENTS; i++) {
21,624✔
94
    if (pVnode->aBufPool[i]) {
16,218!
95
      vnodeBufPoolDestroy(pVnode->aBufPool[i]);
16,218✔
96
      pVnode->aBufPool[i] = NULL;
16,218✔
97
    }
98
  }
99

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

103
void vnodeBufPoolReset(SVBufPool *pPool) {
17,974✔
104
  if (pPool->nQuery != 0) {
17,974!
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) {
47,470✔
110
    pNode->prev->pnext = &pPool->pTail;
29,496✔
111
    pPool->pTail = pNode->prev;
29,496✔
112
    pPool->size = pPool->size - sizeof(*pNode) - pNode->size;
29,496✔
113
    taosMemoryFree(pNode);
29,496✔
114
  }
115

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

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

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

131
  if (pPool->lock) taosThreadSpinLock(pPool->lock);
21,379,127✔
132

133
  ptr = pPool->ptr;
21,390,121✔
134
  paddingLen = (((long)ptr + 7) & ~7) - (long)ptr;
21,390,121✔
135

136
  if (pPool->node.size >= pPool->ptr - pPool->node.data + size + paddingLen) {
21,390,121✔
137
    // allocate from the anchor node
138
    p = pPool->ptr + paddingLen;
21,360,625✔
139
    size += paddingLen;
21,360,625✔
140
    pPool->ptr = pPool->ptr + size;
21,360,625✔
141
    pPool->size += size;
21,360,625✔
142
  } else {
143
    // allocate a new node
144
    pNode = taosMemoryMalloc(sizeof(*pNode) + size);
29,496✔
145
    if (pNode == NULL) {
29,496!
146
      if (pPool->lock) {
×
147
        (void)taosThreadSpinUnlock(pPool->lock);
×
148
      }
149
      return NULL;
×
150
    }
151

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

159
    pPool->size = pPool->size + sizeof(*pNode) + size;
29,496✔
160
  }
161
  if (pPool->lock) (void)taosThreadSpinUnlock(pPool->lock);
21,390,121✔
162
  return p;
21,392,871✔
163
}
164

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

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

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

UNCOV
201
void vnodeBufPoolFree(SVBufPool *pPool, void *p) {
×
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
  // }
UNCOV
213
}
×
214

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

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

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

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

252
  SVnode *pVnode = pPool->pVnode;
7,224✔
253

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

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

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

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

278
  vnodeBufPoolAddToFreeList(pPool);
90✔
279

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

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

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

296
  (void)taosThreadMutexUnlock(&pPool->mutex);
111,141✔
297
}
111,129✔
298

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

302
  if (proactive) {
111,113!
303
    (void)taosThreadMutexLock(&pPool->mutex);
111,120✔
304
  }
305

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

310
  if (proactive) {
111,125✔
311
    (void)taosThreadMutexUnlock(&pPool->mutex);
111,120✔
312
  }
313
}
111,134✔
314

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

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

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

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

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

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

UNCOV
337
_exit:
×
UNCOV
338
  (void)taosThreadMutexUnlock(&pPool->mutex);
×
UNCOV
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