#4143

Committed 24 May 2025 03:30AM UTC coverage: 32.868% (-29.4%) from 62.238%

Build # #4143

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

test: migrate stream cases (#31164)

Run Details

76401 of 312956 branches covered (24.41%)

Branch coverage included in aggregate %.

128686 of 311012 relevant lines covered (41.38%)

579734.08 hits per line

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

48.9

/source/libs/sync/src/syncRaftStore.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/>.
 */

#define _DEFAULT_SOURCE
#include "syncRaftStore.h"
#include "syncUtil.h"
#include "tjson.h"

int32_t raftStoreReadFile(SSyncNode *pNode);
int32_t raftStoreWriteFile(SSyncNode *pNode);

static int32_t raftStoreDecode(const SJson *pJson, SRaftStore *pStore) {
  int32_t code = 0;

  tjsonGetNumberValue(pJson, "current_term", pStore->currentTerm, code);
  if (code < 0) TAOS_RETURN(TSDB_CODE_FAILED);
  tjsonGetNumberValue(pJson, "vote_for_addr", pStore->voteFor.addr, code);
  if (code < 0) TAOS_RETURN(TSDB_CODE_FAILED);
  tjsonGetInt32ValueFromDouble(pJson, "vote_for_vgid", pStore->voteFor.vgId, code);
  if (code < 0) TAOS_RETURN(TSDB_CODE_FAILED);

  TAOS_RETURN(TSDB_CODE_SUCCESS);
}

int32_t raftStoreReadFile(SSyncNode *pNode) {
  int32_t     code = -1, lino = 0;
  TdFilePtr   pFile = NULL;
  char       *pData = NULL;
  SJson      *pJson = NULL;
  const char *file = pNode->raftStorePath;
  SRaftStore *pStore = &pNode->raftStore;

  if (taosStatFile(file, NULL, NULL, NULL) < 0) {
    sInfo("vgId:%d, raft store file:%s not exist, use default value", pNode->vgId, file);
    pStore->currentTerm = 0;
    pStore->voteFor.addr = 0;
    pStore->voteFor.vgId = 0;
    return raftStoreWriteFile(pNode);
  }

  pFile = taosOpenFile(file, TD_FILE_READ);
  if (pFile == NULL) {
    sError("vgId:%d, failed to open raft store file:%s since %s", pNode->vgId, file, terrstr());

    TAOS_CHECK_GOTO(terrno, &lino, _OVER);
  }

  int64_t size = 0;
  code = taosFStatFile(pFile, &size, NULL);
  if (code != 0) {
    sError("vgId:%d, failed to fstat raft store file:%s since %s", pNode->vgId, file, terrstr());
    TAOS_CHECK_GOTO(code, &lino, _OVER);
  }

  pData = taosMemoryMalloc(size + 1);
  if (pData == NULL) {
    TAOS_CHECK_GOTO(terrno, &lino, _OVER);
  }

  if (taosReadFile(pFile, pData, size) != size) {
    sError("vgId:%d, failed to read raft store file:%s since %s", pNode->vgId, file, terrstr());

    TAOS_CHECK_GOTO(terrno, &lino, _OVER);
  }

  pData[size] = '\0';

  pJson = tjsonParse(pData);
  if (pJson == NULL) {
    TAOS_CHECK_GOTO(TSDB_CODE_INVALID_JSON_FORMAT, &lino, _OVER);
  }

  if (raftStoreDecode(pJson, pStore) < 0) {
    TAOS_CHECK_GOTO(TSDB_CODE_INVALID_JSON_FORMAT, &lino, _OVER);
  }

  code = 0;
  sInfo("vgId:%d, succceed to read raft store file %s", pNode->vgId, file);

_OVER:
  if (pData != NULL) taosMemoryFree(pData);
  if (pJson != NULL) cJSON_Delete(pJson);
  if (pFile != NULL) taosCloseFile(&pFile);

  if (code != 0) {
    sError("vgId:%d, failed to read raft store file:%s since %s", pNode->vgId, file, terrstr());
  }

  TAOS_RETURN(code);
}

static int32_t raftStoreEncode(SJson *pJson, SRaftStore *pStore) {
  if (tjsonAddIntegerToObject(pJson, "current_term", pStore->currentTerm) < 0) TAOS_RETURN(TSDB_CODE_FAILED);
  if (tjsonAddIntegerToObject(pJson, "vote_for_addr", pStore->voteFor.addr) < 0) TAOS_RETURN(TSDB_CODE_FAILED);
  if (tjsonAddDoubleToObject(pJson, "vote_for_vgid", pStore->voteFor.vgId) < 0) TAOS_RETURN(TSDB_CODE_FAILED);

  TAOS_RETURN(TSDB_CODE_SUCCESS);
}

int32_t raftStoreWriteFile(SSyncNode *pNode) {
  int32_t     code = -1, lino = 0;
  char       *buffer = NULL;
  SJson      *pJson = NULL;
  TdFilePtr   pFile = NULL;
  const char *realfile = pNode->raftStorePath;
  SRaftStore *pStore = &pNode->raftStore;
  char        file[PATH_MAX] = {0};
  snprintf(file, sizeof(file), "%s.bak", realfile);

  pJson = tjsonCreateObject();
  if (pJson == NULL) TAOS_CHECK_GOTO(terrno, &lino, _OVER);
  if (raftStoreEncode(pJson, pStore) != 0) TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _OVER);

  buffer = tjsonToString(pJson);
  if (buffer == NULL) TAOS_CHECK_GOTO(terrno, &lino, _OVER);

  pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
  if (pFile == NULL) TAOS_CHECK_GOTO(terrno, &lino, _OVER);

  int32_t len = strlen(buffer);
  if (taosWriteFile(pFile, buffer, len) <= 0) TAOS_CHECK_GOTO(terrno, &lino, _OVER);

  if (taosFsyncFile(pFile) < 0) TAOS_CHECK_GOTO(terrno, &lino, _OVER);

  TAOS_CHECK_GOTO(taosCloseFile(&pFile), &lino, _OVER);
  if (taosRenameFile(file, realfile) != 0) TAOS_CHECK_GOTO(terrno, &lino, _OVER);

  code = 0;
  sInfo("vgId:%d, succeed to write raft store file:%s, term:%" PRId64, pNode->vgId, realfile, pStore->currentTerm);

_OVER:
  if (pJson != NULL) tjsonDelete(pJson);
  if (buffer != NULL) taosMemoryFree(buffer);
  if (pFile != NULL) taosCloseFile(&pFile);

  if (code != 0) {
    sError("vgId:%d, failed to write raft store file:%s since %s", pNode->vgId, realfile, terrstr());
  }
  return code;
}

int32_t raftStoreOpen(SSyncNode *pNode) {
  (void)taosThreadMutexInit(&pNode->raftStore.mutex, NULL);
  return raftStoreReadFile(pNode);
}

void raftStoreClose(SSyncNode *pNode) { (void)taosThreadMutexDestroy(&pNode->raftStore.mutex); }

bool raftStoreHasVoted(SSyncNode *pNode) {
  (void)taosThreadMutexLock(&pNode->raftStore.mutex);
  bool b = syncUtilEmptyId(&pNode->raftStore.voteFor);
  (void)taosThreadMutexUnlock(&pNode->raftStore.mutex);
  return (!b);
}

void raftStoreVote(SSyncNode *pNode, SRaftId *pRaftId) {
  (void)taosThreadMutexLock(&pNode->raftStore.mutex);
  pNode->raftStore.voteFor = *pRaftId;
  int32_t code = 0;
  if ((code = raftStoreWriteFile(pNode)) != 0) {
    sError("vgId:%d, failed to write raft store file since %s", pNode->vgId, tstrerror(code));
  }
  (void)taosThreadMutexUnlock(&pNode->raftStore.mutex);
}

void raftStoreClearVote(SSyncNode *pNode) {
  (void)taosThreadMutexLock(&pNode->raftStore.mutex);
  pNode->raftStore.voteFor = EMPTY_RAFT_ID;
  int32_t code = 0;
  if ((code = raftStoreWriteFile(pNode)) != 0) {
    sError("vgId:%d, failed to write raft store file since %s", pNode->vgId, tstrerror(code));
  }
  (void)taosThreadMutexUnlock(&pNode->raftStore.mutex);
}

void raftStoreNextTerm(SSyncNode *pNode) {
  (void)taosThreadMutexLock(&pNode->raftStore.mutex);
  pNode->raftStore.currentTerm++;
  int32_t code = 0;
  if ((code = raftStoreWriteFile(pNode)) != 0) {
    sError("vgId:%d, failed to write raft store file since %s", pNode->vgId, tstrerror(code));
  }
  (void)taosThreadMutexUnlock(&pNode->raftStore.mutex);
}

void raftStoreSetTerm(SSyncNode *pNode, SyncTerm term) {
  (void)taosThreadMutexLock(&pNode->raftStore.mutex);
  if (pNode->raftStore.currentTerm < term) {
    pNode->raftStore.currentTerm = term;
    int32_t code = 0;
    if ((code = raftStoreWriteFile(pNode)) != 0) {
      sError("vgId:%d, failed to write raft store file since %s", pNode->vgId, tstrerror(code));
    }
  }
  (void)taosThreadMutexUnlock(&pNode->raftStore.mutex);
}

SyncTerm raftStoreGetTerm(SSyncNode *pNode) {
  (void)taosThreadMutexLock(&pNode->raftStore.mutex);
  SyncTerm term = pNode->raftStore.currentTerm;
  (void)taosThreadMutexUnlock(&pNode->raftStore.mutex);
  return term;
}

SyncTerm raftStoreTryGetTerm(SSyncNode *pNode) {
  SyncTerm term = 0;
  if (taosThreadMutexTryLock(&pNode->raftStore.mutex) == 0) {
    term = pNode->raftStore.currentTerm;
    (void)taosThreadMutexUnlock(&pNode->raftStore.mutex);
  }

  return term;
}

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	#define _DEFAULT_SOURCE
17	#include "syncRaftStore.h"
18	#include "syncUtil.h"
19	#include "tjson.h"
20
21	int32_t raftStoreReadFile(SSyncNode *pNode);
22	int32_t raftStoreWriteFile(SSyncNode *pNode);
23
24	static int32_t raftStoreDecode(const SJson pJson, SRaftStore pStore) {	15✔
25	int32_t code = 0;	15✔
26
27	tjsonGetNumberValue(pJson, "current_term", pStore->currentTerm, code);	15✔
28	if (code < 0) TAOS_RETURN(TSDB_CODE_FAILED);	15!
29	tjsonGetNumberValue(pJson, "vote_for_addr", pStore->voteFor.addr, code);	15✔
30	if (code < 0) TAOS_RETURN(TSDB_CODE_FAILED);	15!
31	tjsonGetInt32ValueFromDouble(pJson, "vote_for_vgid", pStore->voteFor.vgId, code);	15✔
32	if (code < 0) TAOS_RETURN(TSDB_CODE_FAILED);	15!
33
34	TAOS_RETURN(TSDB_CODE_SUCCESS);	15✔
35	}
36
37	int32_t raftStoreReadFile(SSyncNode *pNode) {	68✔
38	int32_t code = -1, lino = 0;	68✔
39	TdFilePtr pFile = NULL;	68✔
40	char *pData = NULL;	68✔
41	SJson *pJson = NULL;	68✔
42	const char *file = pNode->raftStorePath;	68✔
43	SRaftStore *pStore = &pNode->raftStore;	68✔
44
45	if (taosStatFile(file, NULL, NULL, NULL) < 0) {	68✔
46	sInfo("vgId:%d, raft store file:%s not exist, use default value", pNode->vgId, file);	53✔
47	pStore->currentTerm = 0;	53✔
48	pStore->voteFor.addr = 0;	53✔
49	pStore->voteFor.vgId = 0;	53✔
50	return raftStoreWriteFile(pNode);	53✔
51	}
52
53	pFile = taosOpenFile(file, TD_FILE_READ);	15✔
54	if (pFile == NULL) {	15!
55	sError("vgId:%d, failed to open raft store file:%s since %s", pNode->vgId, file, terrstr());	×
56
57	TAOS_CHECK_GOTO(terrno, &lino, _OVER);	×
58	}
59
60	int64_t size = 0;	15✔
61	code = taosFStatFile(pFile, &size, NULL);	15✔
62	if (code != 0) {	15!
63	sError("vgId:%d, failed to fstat raft store file:%s since %s", pNode->vgId, file, terrstr());	×
64	TAOS_CHECK_GOTO(code, &lino, _OVER);	×
65	}
66
67	pData = taosMemoryMalloc(size + 1);	15!
68	if (pData == NULL) {	15!
69	TAOS_CHECK_GOTO(terrno, &lino, _OVER);	×
70	}
71
72	if (taosReadFile(pFile, pData, size) != size) {	15!
73	sError("vgId:%d, failed to read raft store file:%s since %s", pNode->vgId, file, terrstr());	×
74
75	TAOS_CHECK_GOTO(terrno, &lino, _OVER);	×
76	}
77
78	pData[size] = '\0';	15✔
79
80	pJson = tjsonParse(pData);	15✔
81	if (pJson == NULL) {	15!
82	TAOS_CHECK_GOTO(TSDB_CODE_INVALID_JSON_FORMAT, &lino, _OVER);	×
83	}
84
85	if (raftStoreDecode(pJson, pStore) < 0) {	15!
86	TAOS_CHECK_GOTO(TSDB_CODE_INVALID_JSON_FORMAT, &lino, _OVER);	×
87	}
88
89	code = 0;	15✔
90	sInfo("vgId:%d, succceed to read raft store file %s", pNode->vgId, file);	15!
91
92	_OVER:	×
93	if (pData != NULL) taosMemoryFree(pData);	15!
94	if (pJson != NULL) cJSON_Delete(pJson);	15!
95	if (pFile != NULL) taosCloseFile(&pFile);	15!
96
97	if (code != 0) {	15!
98	sError("vgId:%d, failed to read raft store file:%s since %s", pNode->vgId, file, terrstr());	×
99	}
100
101	TAOS_RETURN(code);	15✔
102	}
103
104	static int32_t raftStoreEncode(SJson pJson, SRaftStore pStore) {	121✔
105	if (tjsonAddIntegerToObject(pJson, "current_term", pStore->currentTerm) < 0) TAOS_RETURN(TSDB_CODE_FAILED);	121!
106	if (tjsonAddIntegerToObject(pJson, "vote_for_addr", pStore->voteFor.addr) < 0) TAOS_RETURN(TSDB_CODE_FAILED);	121!
107	if (tjsonAddDoubleToObject(pJson, "vote_for_vgid", pStore->voteFor.vgId) < 0) TAOS_RETURN(TSDB_CODE_FAILED);	121!
108
109	TAOS_RETURN(TSDB_CODE_SUCCESS);	121✔
110	}
111
112	int32_t raftStoreWriteFile(SSyncNode *pNode) {	121✔
113	int32_t code = -1, lino = 0;	121✔
114	char *buffer = NULL;	121✔
115	SJson *pJson = NULL;	121✔
116	TdFilePtr pFile = NULL;	121✔
117	const char *realfile = pNode->raftStorePath;	121✔
118	SRaftStore *pStore = &pNode->raftStore;	121✔
119	char file[PATH_MAX] = {0};	121✔
120	snprintf(file, sizeof(file), "%s.bak", realfile);	121✔
121
122	pJson = tjsonCreateObject();	121✔
123	if (pJson == NULL) TAOS_CHECK_GOTO(terrno, &lino, _OVER);	121!
124	if (raftStoreEncode(pJson, pStore) != 0) TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _OVER);	121!
125
126	buffer = tjsonToString(pJson);	121✔
127	if (buffer == NULL) TAOS_CHECK_GOTO(terrno, &lino, _OVER);	121!
128
129	pFile = taosOpenFile(file, TD_FILE_CREATE \| TD_FILE_WRITE \| TD_FILE_TRUNC \| TD_FILE_WRITE_THROUGH);	121✔
130	if (pFile == NULL) TAOS_CHECK_GOTO(terrno, &lino, _OVER);	121!
131
132	int32_t len = strlen(buffer);	121✔
133	if (taosWriteFile(pFile, buffer, len) <= 0) TAOS_CHECK_GOTO(terrno, &lino, _OVER);	121!
134
135	if (taosFsyncFile(pFile) < 0) TAOS_CHECK_GOTO(terrno, &lino, _OVER);	121!
136
137	TAOS_CHECK_GOTO(taosCloseFile(&pFile), &lino, _OVER);	121!
138	if (taosRenameFile(file, realfile) != 0) TAOS_CHECK_GOTO(terrno, &lino, _OVER);	121!
139
140	code = 0;	121✔
141	sInfo("vgId:%d, succeed to write raft store file:%s, term:%" PRId64, pNode->vgId, realfile, pStore->currentTerm);	121✔
142
143	_OVER:	56✔
144	if (pJson != NULL) tjsonDelete(pJson);	121!
145	if (buffer != NULL) taosMemoryFree(buffer);	121!
146	if (pFile != NULL) taosCloseFile(&pFile);	121!
147
148	if (code != 0) {	121!
149	sError("vgId:%d, failed to write raft store file:%s since %s", pNode->vgId, realfile, terrstr());	×
150	}
151	return code;	121✔
152	}
153
154	int32_t raftStoreOpen(SSyncNode *pNode) {	68✔
155	(void)taosThreadMutexInit(&pNode->raftStore.mutex, NULL);	68✔
156	return raftStoreReadFile(pNode);	68✔
157	}
158
159	void raftStoreClose(SSyncNode *pNode) { (void)taosThreadMutexDestroy(&pNode->raftStore.mutex); }	68✔
160
161	bool raftStoreHasVoted(SSyncNode *pNode) {	×
162	(void)taosThreadMutexLock(&pNode->raftStore.mutex);	×
163	bool b = syncUtilEmptyId(&pNode->raftStore.voteFor);	×
164	(void)taosThreadMutexUnlock(&pNode->raftStore.mutex);	×
165	return (!b);	×
166	}
167
168	void raftStoreVote(SSyncNode pNode, SRaftId pRaftId) {	×
169	(void)taosThreadMutexLock(&pNode->raftStore.mutex);	×
170	pNode->raftStore.voteFor = *pRaftId;	×
171	int32_t code = 0;	×
172	if ((code = raftStoreWriteFile(pNode)) != 0) {	×
173	sError("vgId:%d, failed to write raft store file since %s", pNode->vgId, tstrerror(code));	×
174	}
175	(void)taosThreadMutexUnlock(&pNode->raftStore.mutex);	×
176	}	×
177
178	void raftStoreClearVote(SSyncNode *pNode) {	×
179	(void)taosThreadMutexLock(&pNode->raftStore.mutex);	×
180	pNode->raftStore.voteFor = EMPTY_RAFT_ID;	×
181	int32_t code = 0;	×
182	if ((code = raftStoreWriteFile(pNode)) != 0) {	×
183	sError("vgId:%d, failed to write raft store file since %s", pNode->vgId, tstrerror(code));	×
184	}
185	(void)taosThreadMutexUnlock(&pNode->raftStore.mutex);	×
186	}	×
187
188	void raftStoreNextTerm(SSyncNode *pNode) {	68✔
189	(void)taosThreadMutexLock(&pNode->raftStore.mutex);	68✔
190	pNode->raftStore.currentTerm++;	68✔
191	int32_t code = 0;	68✔
192	if ((code = raftStoreWriteFile(pNode)) != 0) {	68!
193	sError("vgId:%d, failed to write raft store file since %s", pNode->vgId, tstrerror(code));	×
194	}
195	(void)taosThreadMutexUnlock(&pNode->raftStore.mutex);	68✔
196	}	68✔
197
198	void raftStoreSetTerm(SSyncNode *pNode, SyncTerm term) {	×
199	(void)taosThreadMutexLock(&pNode->raftStore.mutex);	×
200	if (pNode->raftStore.currentTerm < term) {	×
201	pNode->raftStore.currentTerm = term;	×
202	int32_t code = 0;	×
203	if ((code = raftStoreWriteFile(pNode)) != 0) {	×
204	sError("vgId:%d, failed to write raft store file since %s", pNode->vgId, tstrerror(code));	×
205	}
206	}
207	(void)taosThreadMutexUnlock(&pNode->raftStore.mutex);	×
208	}	×
209
210	SyncTerm raftStoreGetTerm(SSyncNode *pNode) {	85,008✔
211	(void)taosThreadMutexLock(&pNode->raftStore.mutex);	85,008✔
212	SyncTerm term = pNode->raftStore.currentTerm;	85,009✔
213	(void)taosThreadMutexUnlock(&pNode->raftStore.mutex);	85,009✔
214	return term;	85,009✔
215	}
216
217	SyncTerm raftStoreTryGetTerm(SSyncNode *pNode) {	×
218	SyncTerm term = 0;	×
219	if (taosThreadMutexTryLock(&pNode->raftStore.mutex) == 0) {	×
220	term = pNode->raftStore.currentTerm;	×
221	(void)taosThreadMutexUnlock(&pNode->raftStore.mutex);	×
222	}
223
224	return term;	×
225	}

taosdata / TDengine / #4143

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