#3599

Committed 08 Feb 2025 11:23AM UTC coverage: 1.77% (-61.6%) from 63.396%

Build # #3599

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #29712 from taosdata/fix/TD-33652-3.0

fix: reduce write rows from 30w to 3w

Run Details

3776 of 278949 branches covered (1.35%)

Branch coverage included in aggregate %.

6012 of 274147 relevant lines covered (2.19%)

1642.73 hits per line

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

0.0

/source/dnode/vnode/src/tq/tqStreamTaskSnap.c

/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#include "meta.h"
#include "tdbInt.h"
#include "tq.h"

// STqSnapReader ========================================

typedef struct {
  int8_t type;
  TTB*   tbl;
} STablePair;
struct SStreamTaskReader {
  STQ*    pTq;
  int64_t sver;
  int64_t ever;
  TBC*    pCur;
  SArray* tdbTbList;
  int8_t  pos;
};

int32_t streamTaskSnapReaderOpen(STQ* pTq, int64_t sver, int64_t ever, SStreamTaskReader** ppReader) {
  int32_t            code = 0;
  SStreamTaskReader* pReader = NULL;

  // alloc
  pReader = (SStreamTaskReader*)taosMemoryCalloc(1, sizeof(SStreamTaskReader));
  if (pReader == NULL) {
    TAOS_CHECK_RETURN(terrno);
  }
  pReader->pTq = pTq;
  pReader->sver = sver;
  pReader->ever = ever;
  pReader->tdbTbList = taosArrayInit(4, sizeof(STablePair));
  if (pReader->tdbTbList == NULL) {
    TAOS_CHECK_GOTO(terrno, NULL, _err);
  }

  STablePair pair1 = {.tbl = pTq->pStreamMeta->pTaskDb, .type = SNAP_DATA_STREAM_TASK};
  if (NULL == taosArrayPush(pReader->tdbTbList, &pair1)) {
    TAOS_CHECK_GOTO(terrno, NULL, _err);
  }

  STablePair pair2 = {.tbl = pTq->pStreamMeta->pCheckpointDb, .type = SNAP_DATA_STREAM_TASK_CHECKPOINT};
  if (NULL == taosArrayPush(pReader->tdbTbList, &pair2)) {
    TAOS_CHECK_GOTO(terrno, NULL, _err);
  }

  pReader->pos = 0;

  STablePair* pPair = taosArrayGet(pReader->tdbTbList, pReader->pos);
  code = tdbTbcOpen(pPair->tbl, &pReader->pCur, NULL);
  if (code) {
    tqInfo("vgId:%d, vnode stream-task snapshot reader failed to open, reason: %s", TD_VID(pTq->pVnode),
           tstrerror(code));
    TAOS_CHECK_GOTO(code, NULL, _err);
  }

  code = tdbTbcMoveToFirst(pReader->pCur);
  if (code) {
    tqInfo("vgId:%d, vnode stream-task snapshot reader failed to iterate, reason: %s", TD_VID(pTq->pVnode),
           tstrerror(code));
    TAOS_CHECK_GOTO(code, NULL, _err);
  }

  tqDebug("vgId:%d, vnode stream-task snapshot reader opened", TD_VID(pTq->pVnode));

  *ppReader = pReader;
  return code;

_err:
  tqError("vgId:%d, vnode stream-task snapshot reader open failed since %s", TD_VID(pTq->pVnode), tstrerror(code));
  int32_t ret = streamTaskSnapReaderClose(pReader);
  *ppReader = NULL;
  return code;
}

int32_t streamTaskSnapReaderClose(SStreamTaskReader* pReader) {
  if (pReader == NULL) {
    return 0;
  }

  int32_t code = 0;
  int32_t vgId = TD_VID(pReader->pTq->pVnode);

  taosArrayDestroy(pReader->tdbTbList);
  tdbTbcClose(pReader->pCur);
  tqInfo("vgId:%d, vnode stream-task snapshot reader closed", vgId);

  taosMemoryFree(pReader);
  return code;
}

int32_t streamTaskSnapRead(SStreamTaskReader* pReader, uint8_t** ppData) {
  int32_t     code = 0;
  const void* pKey = NULL;
  void*       pVal = NULL;
  int32_t     kLen = 0;
  int32_t     vLen = 0;
  SDecoder    decoder;
  STqHandle   handle;

  *ppData = NULL;
  int8_t except = 0;
  tqDebug("vgId:%d, vnode stream-task snapshot start read data", TD_VID(pReader->pTq->pVnode));

  STablePair* pPair = taosArrayGet(pReader->tdbTbList, pReader->pos);

NextTbl:
  except = 0;
  for (;;) {
    const void* tVal = NULL;
    int32_t     tLen = 0;
    if (tdbTbcGet(pReader->pCur, &pKey, &kLen, &tVal, &tLen)) {
      except = 1;
      break;
    } else {
      pVal = taosMemoryCalloc(1, tLen);
      if (pVal == NULL) {
        code = terrno;
        goto _err;
      }

      memcpy(pVal, tVal, tLen);
      vLen = tLen;
    }
    TAOS_UNUSED(tdbTbcMoveToNext(pReader->pCur));
    break;
  }
  if (except == 1) {
    if (pReader->pos + 1 < taosArrayGetSize(pReader->tdbTbList)) {
      tdbTbcClose(pReader->pCur);

      pReader->pos += 1;
      pPair = taosArrayGet(pReader->tdbTbList, pReader->pos);
      code = tdbTbcOpen(pPair->tbl, &pReader->pCur, NULL);
      TAOS_UNUSED(tdbTbcMoveToFirst(pReader->pCur));

      goto NextTbl;
    }
  }
  if (pVal == NULL || vLen == 0) {
    *ppData = NULL;
    tqDebug("vgId:%d, vnode stream-task snapshot finished read data", TD_VID(pReader->pTq->pVnode));
    return code;
  }
  *ppData = taosMemoryMalloc(sizeof(SSnapDataHdr) + vLen);
  if (*ppData == NULL) {
    code = terrno;
    goto _err;
  }

  SSnapDataHdr* pHdr = (SSnapDataHdr*)(*ppData);
  pHdr->type = pPair->type;
  pHdr->size = vLen;
  memcpy(pHdr->data, pVal, vLen);
  taosMemoryFree(pVal);

  tqDebug("vgId:%d, vnode stream-task snapshot read data vLen:%d", TD_VID(pReader->pTq->pVnode), vLen);
  return code;

_err:
  tqError("vgId:%d, vnode stream-task snapshot read data failed since %s", TD_VID(pReader->pTq->pVnode),
          tstrerror(code));
  return code;
}

// STqSnapWriter ========================================
struct SStreamTaskWriter {
  STQ*    pTq;
  int64_t sver;
  int64_t ever;
};

int32_t streamTaskSnapWriterOpen(STQ* pTq, int64_t sver, int64_t ever, SStreamTaskWriter** ppWriter) {
  int32_t            code = 0;
  SStreamTaskWriter* pWriter;

  // alloc
  pWriter = (SStreamTaskWriter*)taosMemoryCalloc(1, sizeof(*pWriter));
  if (pWriter == NULL) {
    TAOS_CHECK_RETURN(terrno);
  }
  pWriter->pTq = pTq;
  pWriter->sver = sver;
  pWriter->ever = ever;

  *ppWriter = pWriter;
  tqDebug("vgId:%d, vnode stream-task snapshot writer opened", TD_VID(pTq->pVnode));
  return code;
}

int32_t streamTaskSnapWriterClose(SStreamTaskWriter* pWriter, int8_t rollback, int8_t loadTask) {
  int32_t code = 0;
  STQ*    pTq = pWriter->pTq;

  streamMetaWLock(pTq->pStreamMeta);
  tqDebug("vgId:%d, vnode stream-task snapshot writer closed", TD_VID(pTq->pVnode));
  if (rollback) {
    TAOS_UNUSED(tdbAbort(pTq->pStreamMeta->db, pTq->pStreamMeta->txn));
  } else {
    code = tdbCommit(pTq->pStreamMeta->db, pTq->pStreamMeta->txn);
    if (code) goto _err;
    code = tdbPostCommit(pTq->pStreamMeta->db, pTq->pStreamMeta->txn);
    if (code) goto _err;
  }

  if ((code = tdbBegin(pTq->pStreamMeta->db, &pTq->pStreamMeta->txn, tdbDefaultMalloc, tdbDefaultFree, NULL, 0)) < 0) {
    taosMemoryFree(pWriter);
    goto _err;
  }
  streamMetaWUnLock(pTq->pStreamMeta);
  taosMemoryFree(pWriter);

  if (loadTask == 1) {
    streamMetaLoadAllTasks(pTq->pStreamMeta);
  }
  return code;

_err:
  tqError("vgId:%d, vnode stream-task snapshot writer failed to close since %s", TD_VID(pWriter->pTq->pVnode),
          tstrerror(code));
  streamMetaWUnLock(pTq->pStreamMeta);
  return code;
}

int32_t streamTaskSnapWrite(SStreamTaskWriter* pWriter, uint8_t* pData, uint32_t nData) {
  int32_t       code = 0;
  STQ*          pTq = pWriter->pTq;
  SSnapDataHdr* pHdr = (SSnapDataHdr*)pData;
  if (pHdr->type == SNAP_DATA_STREAM_TASK) {
    STaskId taskId = {0};

    SDecoder decoder;
    tDecoderInit(&decoder, (uint8_t*)pData + sizeof(SSnapDataHdr), nData - sizeof(SSnapDataHdr));
    code = tDecodeStreamTaskId(&decoder, &taskId);
    if (code < 0) {
      tDecoderClear(&decoder);
      goto _err;
    }
    tDecoderClear(&decoder);

    int64_t key[2] = {taskId.streamId, taskId.taskId};
    streamMetaWLock(pTq->pStreamMeta);
    if ((code =
             tdbTbUpsert(pTq->pStreamMeta->pTaskDb, key, sizeof(int64_t) << 1, (uint8_t*)pData + sizeof(SSnapDataHdr),
                         nData - sizeof(SSnapDataHdr), pTq->pStreamMeta->txn)) < 0) {
      streamMetaWUnLock(pTq->pStreamMeta);
      return code;
    }
    streamMetaWUnLock(pTq->pStreamMeta);
  } else if (pHdr->type == SNAP_DATA_STREAM_TASK_CHECKPOINT) {
    // do nothing
  }
  tqDebug("vgId:%d, vnode stream-task snapshot write", TD_VID(pTq->pVnode));

  return code;

_err:
  tqError("vgId:%d, vnode stream-task snapshot failed to write since %s", TD_VID(pTq->pVnode), tstrerror(code));
  return code;
}

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 "meta.h"
17	#include "tdbInt.h"
18	#include "tq.h"
19
20	// STqSnapReader ========================================
21
22	typedef struct {
23	int8_t type;
24	TTB* tbl;
25	} STablePair;
26	struct SStreamTaskReader {
27	STQ* pTq;
28	int64_t sver;
29	int64_t ever;
30	TBC* pCur;
31	SArray* tdbTbList;
32	int8_t pos;
33	};
34
35	int32_t streamTaskSnapReaderOpen(STQ* pTq, int64_t sver, int64_t ever, SStreamTaskReader** ppReader) {	×
36	int32_t code = 0;	×
37	SStreamTaskReader* pReader = NULL;	×
38
39	// alloc
40	pReader = (SStreamTaskReader*)taosMemoryCalloc(1, sizeof(SStreamTaskReader));	×
41	if (pReader == NULL) {	×
42	TAOS_CHECK_RETURN(terrno);	×
43	}
44	pReader->pTq = pTq;	×
45	pReader->sver = sver;	×
46	pReader->ever = ever;	×
47	pReader->tdbTbList = taosArrayInit(4, sizeof(STablePair));	×
48	if (pReader->tdbTbList == NULL) {	×
49	TAOS_CHECK_GOTO(terrno, NULL, _err);	×
50	}
51
52	STablePair pair1 = {.tbl = pTq->pStreamMeta->pTaskDb, .type = SNAP_DATA_STREAM_TASK};	×
53	if (NULL == taosArrayPush(pReader->tdbTbList, &pair1)) {	×
54	TAOS_CHECK_GOTO(terrno, NULL, _err);	×
55	}
56
57	STablePair pair2 = {.tbl = pTq->pStreamMeta->pCheckpointDb, .type = SNAP_DATA_STREAM_TASK_CHECKPOINT};	×
58	if (NULL == taosArrayPush(pReader->tdbTbList, &pair2)) {	×
59	TAOS_CHECK_GOTO(terrno, NULL, _err);	×
60	}
61
62	pReader->pos = 0;	×
63
64	STablePair* pPair = taosArrayGet(pReader->tdbTbList, pReader->pos);	×
65	code = tdbTbcOpen(pPair->tbl, &pReader->pCur, NULL);	×
66	if (code) {	×
67	tqInfo("vgId:%d, vnode stream-task snapshot reader failed to open, reason: %s", TD_VID(pTq->pVnode),	×
68	tstrerror(code));
69	TAOS_CHECK_GOTO(code, NULL, _err);	×
70	}
71
72	code = tdbTbcMoveToFirst(pReader->pCur);	×
73	if (code) {	×
74	tqInfo("vgId:%d, vnode stream-task snapshot reader failed to iterate, reason: %s", TD_VID(pTq->pVnode),	×
75	tstrerror(code));
76	TAOS_CHECK_GOTO(code, NULL, _err);	×
77	}
78
79	tqDebug("vgId:%d, vnode stream-task snapshot reader opened", TD_VID(pTq->pVnode));	×
80
81	*ppReader = pReader;	×
82	return code;	×
83
84	_err:	×
85	tqError("vgId:%d, vnode stream-task snapshot reader open failed since %s", TD_VID(pTq->pVnode), tstrerror(code));	×
86	int32_t ret = streamTaskSnapReaderClose(pReader);	×
87	*ppReader = NULL;	×
88	return code;	×
89	}
90
91	int32_t streamTaskSnapReaderClose(SStreamTaskReader* pReader) {	×
92	if (pReader == NULL) {	×
93	return 0;	×
94	}
95
96	int32_t code = 0;	×
97	int32_t vgId = TD_VID(pReader->pTq->pVnode);	×
98
99	taosArrayDestroy(pReader->tdbTbList);	×
100	tdbTbcClose(pReader->pCur);	×
101	tqInfo("vgId:%d, vnode stream-task snapshot reader closed", vgId);	×
102
103	taosMemoryFree(pReader);	×
104	return code;	×
105	}
106
107	int32_t streamTaskSnapRead(SStreamTaskReader* pReader, uint8_t** ppData) {	×
108	int32_t code = 0;	×
109	const void* pKey = NULL;	×
110	void* pVal = NULL;	×
111	int32_t kLen = 0;	×
112	int32_t vLen = 0;	×
113	SDecoder decoder;
114	STqHandle handle;
115
116	*ppData = NULL;	×
117	int8_t except = 0;	×
118	tqDebug("vgId:%d, vnode stream-task snapshot start read data", TD_VID(pReader->pTq->pVnode));	×
119
120	STablePair* pPair = taosArrayGet(pReader->tdbTbList, pReader->pos);	×
121
122	NextTbl:	×
123	except = 0;	×
124	for (;;) {
125	const void* tVal = NULL;	×
126	int32_t tLen = 0;	×
127	if (tdbTbcGet(pReader->pCur, &pKey, &kLen, &tVal, &tLen)) {	×
128	except = 1;	×
129	break;	×
130	} else {
131	pVal = taosMemoryCalloc(1, tLen);	×
132	if (pVal == NULL) {	×
133	code = terrno;	×
134	goto _err;	×
135	}
136
137	memcpy(pVal, tVal, tLen);	×
138	vLen = tLen;	×
139	}
140	TAOS_UNUSED(tdbTbcMoveToNext(pReader->pCur));	×
141	break;	×
142	}
143	if (except == 1) {	×
144	if (pReader->pos + 1 < taosArrayGetSize(pReader->tdbTbList)) {	×
145	tdbTbcClose(pReader->pCur);	×
146
147	pReader->pos += 1;	×
148	pPair = taosArrayGet(pReader->tdbTbList, pReader->pos);	×
149	code = tdbTbcOpen(pPair->tbl, &pReader->pCur, NULL);	×
150	TAOS_UNUSED(tdbTbcMoveToFirst(pReader->pCur));	×
151
152	goto NextTbl;	×
153	}
154	}
155	if (pVal == NULL \|\| vLen == 0) {	×
156	*ppData = NULL;	×
157	tqDebug("vgId:%d, vnode stream-task snapshot finished read data", TD_VID(pReader->pTq->pVnode));	×
158	return code;	×
159	}
160	*ppData = taosMemoryMalloc(sizeof(SSnapDataHdr) + vLen);	×
161	if (*ppData == NULL) {	×
162	code = terrno;	×
163	goto _err;	×
164	}
165
166	SSnapDataHdr* pHdr = (SSnapDataHdr)(ppData);	×
167	pHdr->type = pPair->type;	×
168	pHdr->size = vLen;	×
169	memcpy(pHdr->data, pVal, vLen);	×
170	taosMemoryFree(pVal);	×
171
172	tqDebug("vgId:%d, vnode stream-task snapshot read data vLen:%d", TD_VID(pReader->pTq->pVnode), vLen);	×
173	return code;	×
174
175	_err:	×
176	tqError("vgId:%d, vnode stream-task snapshot read data failed since %s", TD_VID(pReader->pTq->pVnode),	×
177	tstrerror(code));
178	return code;	×
179	}
180
181	// STqSnapWriter ========================================
182	struct SStreamTaskWriter {
183	STQ* pTq;
184	int64_t sver;
185	int64_t ever;
186	};
187
188	int32_t streamTaskSnapWriterOpen(STQ* pTq, int64_t sver, int64_t ever, SStreamTaskWriter** ppWriter) {	×
189	int32_t code = 0;	×
190	SStreamTaskWriter* pWriter;
191
192	// alloc
193	pWriter = (SStreamTaskWriter)taosMemoryCalloc(1, sizeof(pWriter));	×
194	if (pWriter == NULL) {	×
195	TAOS_CHECK_RETURN(terrno);	×
196	}
197	pWriter->pTq = pTq;	×
198	pWriter->sver = sver;	×
199	pWriter->ever = ever;	×
200
201	*ppWriter = pWriter;	×
202	tqDebug("vgId:%d, vnode stream-task snapshot writer opened", TD_VID(pTq->pVnode));	×
203	return code;	×
204	}
205
206	int32_t streamTaskSnapWriterClose(SStreamTaskWriter* pWriter, int8_t rollback, int8_t loadTask) {	×
207	int32_t code = 0;	×
208	STQ* pTq = pWriter->pTq;	×
209
210	streamMetaWLock(pTq->pStreamMeta);	×
211	tqDebug("vgId:%d, vnode stream-task snapshot writer closed", TD_VID(pTq->pVnode));	×
212	if (rollback) {	×
213	TAOS_UNUSED(tdbAbort(pTq->pStreamMeta->db, pTq->pStreamMeta->txn));	×
214	} else {
215	code = tdbCommit(pTq->pStreamMeta->db, pTq->pStreamMeta->txn);	×
216	if (code) goto _err;	×
217	code = tdbPostCommit(pTq->pStreamMeta->db, pTq->pStreamMeta->txn);	×
218	if (code) goto _err;	×
219	}
220
221	if ((code = tdbBegin(pTq->pStreamMeta->db, &pTq->pStreamMeta->txn, tdbDefaultMalloc, tdbDefaultFree, NULL, 0)) < 0) {	×
222	taosMemoryFree(pWriter);	×
223	goto _err;	×
224	}
225	streamMetaWUnLock(pTq->pStreamMeta);	×
226	taosMemoryFree(pWriter);	×
227
228	if (loadTask == 1) {	×
229	streamMetaLoadAllTasks(pTq->pStreamMeta);	×
230	}
231	return code;	×
232
233	_err:	×
234	tqError("vgId:%d, vnode stream-task snapshot writer failed to close since %s", TD_VID(pWriter->pTq->pVnode),	×
235	tstrerror(code));
236	streamMetaWUnLock(pTq->pStreamMeta);	×
237	return code;	×
238	}
239
240	int32_t streamTaskSnapWrite(SStreamTaskWriter* pWriter, uint8_t* pData, uint32_t nData) {	×
241	int32_t code = 0;	×
242	STQ* pTq = pWriter->pTq;	×
243	SSnapDataHdr* pHdr = (SSnapDataHdr*)pData;	×
244	if (pHdr->type == SNAP_DATA_STREAM_TASK) {	×
245	STaskId taskId = {0};	×
246
247	SDecoder decoder;
248	tDecoderInit(&decoder, (uint8_t*)pData + sizeof(SSnapDataHdr), nData - sizeof(SSnapDataHdr));	×
249	code = tDecodeStreamTaskId(&decoder, &taskId);	×
250	if (code < 0) {	×
251	tDecoderClear(&decoder);	×
252	goto _err;	×
253	}
254	tDecoderClear(&decoder);	×
255
256	int64_t key[2] = {taskId.streamId, taskId.taskId};	×
257	streamMetaWLock(pTq->pStreamMeta);	×
258	if ((code =	×
259	tdbTbUpsert(pTq->pStreamMeta->pTaskDb, key, sizeof(int64_t) << 1, (uint8_t*)pData + sizeof(SSnapDataHdr),	×
260	nData - sizeof(SSnapDataHdr), pTq->pStreamMeta->txn)) < 0) {	×
261	streamMetaWUnLock(pTq->pStreamMeta);	×
262	return code;	×
263	}
264	streamMetaWUnLock(pTq->pStreamMeta);	×
265	} else if (pHdr->type == SNAP_DATA_STREAM_TASK_CHECKPOINT) {	×
266	// do nothing
267	}
268	tqDebug("vgId:%d, vnode stream-task snapshot write", TD_VID(pTq->pVnode));	×
269
270	return code;	×
271
272	_err:	×
273	tqError("vgId:%d, vnode stream-task snapshot failed to write since %s", TD_VID(pTq->pVnode), tstrerror(code));	×
274	return code;	×
275	}

taosdata / TDengine / #3599

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

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