#4308

Committed 14 Jun 2025 02:06PM UTC coverage: 62.454% (-0.3%) from 62.777%

Build # #4308

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

fix: taosdump windows pthread_mutex_unlock crash(3.0) (#31357)

* fix: windows pthread_mutex_unlock crash

* enh: sync from main fix taosdump crash windows

* fix: restore .github action branch to main

Run Details

153985 of 315105 branches covered (48.87%)

Branch coverage included in aggregate %.

238120 of 312727 relevant lines covered (76.14%)

6462519.65 hits per line

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

63.6

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

taosdata / TDengine / #4308

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