#3638

Committed 11 Mar 2025 12:59PM UTC coverage: 3.066% (-18.3%) from 21.409%

Build # #3638

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #30118 from taosdata/wl30

udpate ci workflow

Run Details

5914 of 287117 branches covered (2.06%)

Branch coverage included in aggregate %.

11588 of 283747 relevant lines covered (4.08%)

142.17 hits per line

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

0.0

/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) {	×
25	int32_t code = 0;	×
26
27	tjsonGetNumberValue(pJson, "current_term", pStore->currentTerm, code);	×
28	if (code < 0) TAOS_RETURN(TSDB_CODE_FAILED);	×
29	tjsonGetNumberValue(pJson, "vote_for_addr", pStore->voteFor.addr, code);	×
30	if (code < 0) TAOS_RETURN(TSDB_CODE_FAILED);	×
31	tjsonGetInt32ValueFromDouble(pJson, "vote_for_vgid", pStore->voteFor.vgId, code);	×
32	if (code < 0) TAOS_RETURN(TSDB_CODE_FAILED);	×
33
34	TAOS_RETURN(TSDB_CODE_SUCCESS);	×
35	}
36
37	int32_t raftStoreReadFile(SSyncNode *pNode) {	×
38	int32_t code = -1, lino = 0;	×
39	TdFilePtr pFile = NULL;	×
40	char *pData = NULL;	×
41	SJson *pJson = NULL;	×
42	const char *file = pNode->raftStorePath;	×
43	SRaftStore *pStore = &pNode->raftStore;	×
44
45	if (taosStatFile(file, NULL, NULL, NULL) < 0) {	×
46	sInfo("vgId:%d, raft store file:%s not exist, use default value", pNode->vgId, file);	×
47	pStore->currentTerm = 0;	×
48	pStore->voteFor.addr = 0;	×
49	pStore->voteFor.vgId = 0;	×
50	return raftStoreWriteFile(pNode);	×
51	}
52
53	pFile = taosOpenFile(file, TD_FILE_READ);	×
54	if (pFile == NULL) {	×
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;	×
61	code = taosFStatFile(pFile, &size, NULL);	×
62	if (code != 0) {	×
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);	×
68	if (pData == NULL) {	×
69	TAOS_CHECK_GOTO(terrno, &lino, _OVER);	×
70	}
71
72	if (taosReadFile(pFile, pData, size) != size) {	×
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';	×
79
80	pJson = tjsonParse(pData);	×
81	if (pJson == NULL) {	×
82	TAOS_CHECK_GOTO(TSDB_CODE_INVALID_JSON_FORMAT, &lino, _OVER);	×
83	}
84
85	if (raftStoreDecode(pJson, pStore) < 0) {	×
86	TAOS_CHECK_GOTO(TSDB_CODE_INVALID_JSON_FORMAT, &lino, _OVER);	×
87	}
88
89	code = 0;	×
90	sInfo("vgId:%d, succceed to read raft store file %s", pNode->vgId, file);	×
91
92	_OVER:	×
93	if (pData != NULL) taosMemoryFree(pData);	×
94	if (pJson != NULL) cJSON_Delete(pJson);	×
95	if (pFile != NULL) taosCloseFile(&pFile);	×
96
97	if (code != 0) {	×
98	sError("vgId:%d, failed to read raft store file:%s since %s", pNode->vgId, file, terrstr());	×
99	}
100
101	TAOS_RETURN(code);	×
102	}
103
104	static int32_t raftStoreEncode(SJson pJson, SRaftStore pStore) {	×
105	if (tjsonAddIntegerToObject(pJson, "current_term", pStore->currentTerm) < 0) TAOS_RETURN(TSDB_CODE_FAILED);	×
106	if (tjsonAddIntegerToObject(pJson, "vote_for_addr", pStore->voteFor.addr) < 0) TAOS_RETURN(TSDB_CODE_FAILED);	×
107	if (tjsonAddDoubleToObject(pJson, "vote_for_vgid", pStore->voteFor.vgId) < 0) TAOS_RETURN(TSDB_CODE_FAILED);	×
108
109	TAOS_RETURN(TSDB_CODE_SUCCESS);	×
110	}
111
112	int32_t raftStoreWriteFile(SSyncNode *pNode) {	×
113	int32_t code = -1, lino = 0;	×
114	char *buffer = NULL;	×
115	SJson *pJson = NULL;	×
116	TdFilePtr pFile = NULL;	×
117	const char *realfile = pNode->raftStorePath;	×
118	SRaftStore *pStore = &pNode->raftStore;	×
119	char file[PATH_MAX] = {0};	×
120	snprintf(file, sizeof(file), "%s.bak", realfile);	×
121
122	pJson = tjsonCreateObject();	×
123	if (pJson == NULL) TAOS_CHECK_GOTO(terrno, &lino, _OVER);	×
124	if (raftStoreEncode(pJson, pStore) != 0) TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _OVER);	×
125
126	buffer = tjsonToString(pJson);	×
127	if (buffer == NULL) TAOS_CHECK_GOTO(terrno, &lino, _OVER);	×
128
129	pFile = taosOpenFile(file, TD_FILE_CREATE \| TD_FILE_WRITE \| TD_FILE_TRUNC \| TD_FILE_WRITE_THROUGH);	×
130	if (pFile == NULL) TAOS_CHECK_GOTO(terrno, &lino, _OVER);	×
131
132	int32_t len = strlen(buffer);	×
133	if (taosWriteFile(pFile, buffer, len) <= 0) TAOS_CHECK_GOTO(terrno, &lino, _OVER);	×
134
135	if (taosFsyncFile(pFile) < 0) TAOS_CHECK_GOTO(terrno, &lino, _OVER);	×
136
137	TAOS_CHECK_GOTO(taosCloseFile(&pFile), &lino, _OVER);	×
138	if (taosRenameFile(file, realfile) != 0) TAOS_CHECK_GOTO(terrno, &lino, _OVER);	×
139
140	code = 0;	×
141	sInfo("vgId:%d, succeed to write raft store file:%s, term:%" PRId64, pNode->vgId, realfile, pStore->currentTerm);	×
142
143	_OVER:	×
144	if (pJson != NULL) tjsonDelete(pJson);	×
145	if (buffer != NULL) taosMemoryFree(buffer);	×
146	if (pFile != NULL) taosCloseFile(&pFile);	×
147
148	if (code != 0) {	×
149	sError("vgId:%d, failed to write raft store file:%s since %s", pNode->vgId, realfile, terrstr());	×
150	}
151	return code;	×
152	}
153
154	int32_t raftStoreOpen(SSyncNode *pNode) {	×
155	(void)taosThreadMutexInit(&pNode->raftStore.mutex, NULL);	×
156	return raftStoreReadFile(pNode);	×
157	}
158
159	void raftStoreClose(SSyncNode *pNode) { (void)taosThreadMutexDestroy(&pNode->raftStore.mutex); }	×
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) {	×
189	(void)taosThreadMutexLock(&pNode->raftStore.mutex);	×
190	pNode->raftStore.currentTerm++;	×
191	int32_t code = 0;	×
192	if ((code = raftStoreWriteFile(pNode)) != 0) {	×
193	sError("vgId:%d, failed to write raft store file since %s", pNode->vgId, tstrerror(code));	×
194	}
195	(void)taosThreadMutexUnlock(&pNode->raftStore.mutex);	×
196	}	×
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) {	×
211	(void)taosThreadMutexLock(&pNode->raftStore.mutex);	×
212	SyncTerm term = pNode->raftStore.currentTerm;	×
213	(void)taosThreadMutexUnlock(&pNode->raftStore.mutex);	×
214	return term;	×
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 / #3638

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