#3527

Committed 12 Nov 2024 08:50PM UTC coverage: 60.819% (+0.6%) from 60.225%

Build # #3527

Build Type

push

travis-ci

Committed by GitHub

Commit Message

Merge pull request #28730 from taosdata/doc/analysis

Run Details

118574 of 249004 branches covered (47.62%)

Branch coverage included in aggregate %.

199137 of 273386 relevant lines covered (72.84%)

15792295.89 hits per line

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

61.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;
}

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

taosdata / TDengine / #3527

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