#3562

Committed 20 Dec 2024 09:57AM UTC coverage: 26.655% (-32.2%) from 58.812%

Build # #3562

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #29229 from taosdata/enh/TS-5749-3.0

enh: seperate tsdb async tasks to different thread pools

Run Details

21498 of 109421 branches covered (19.65%)

Branch coverage included in aggregate %.

66 of 96 new or added lines in 7 files covered. (68.75%)

39441 existing lines in 157 files now uncovered.

35007 of 102566 relevant lines covered (34.13%)

53922.97 hits per line

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

43.42

/source/dnode/mgmt/mgmt_mnode/src/mmFile.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 "mmInt.h"
#include "tjson.h"

static int32_t mmDecodeOption(SJson *pJson, SMnodeOpt *pOption) {
  int32_t code = 0;

  tjsonGetInt32ValueFromDouble(pJson, "deployed", pOption->deploy, code);
  if (code < 0) return code;
  tjsonGetInt32ValueFromDouble(pJson, "selfIndex", pOption->selfIndex, code);
  if (code < 0) return code;
  tjsonGetInt32ValueFromDouble(pJson, "lastIndex", pOption->lastIndex, code);
  if (code < 0) return code;

  SJson *replicas = tjsonGetObjectItem(pJson, "replicas");
  if (replicas == NULL) return 0;
  pOption->numOfTotalReplicas = tjsonGetArraySize(replicas);

  pOption->numOfReplicas = 0;

  for (int32_t i = 0; i < pOption->numOfTotalReplicas; ++i) {
    SJson *replica = tjsonGetArrayItem(replicas, i);
    if (replica == NULL) return TSDB_CODE_INVALID_JSON_FORMAT;

    SReplica *pReplica = pOption->replicas + i;
    tjsonGetInt32ValueFromDouble(replica, "id", pReplica->id, code);
    if (code < 0) return code;
    code = tjsonGetStringValue(replica, "fqdn", pReplica->fqdn);
    if (code < 0) return code;
    tjsonGetUInt16ValueFromDouble(replica, "port", pReplica->port, code);
    if (code < 0) return code;
    tjsonGetInt32ValueFromDouble(replica, "role", pOption->nodeRoles[i], code);
    if (code < 0) return code;
    if (pOption->nodeRoles[i] == TAOS_SYNC_ROLE_VOTER) {
      pOption->numOfReplicas++;
    }
  }

  for (int32_t i = 0; i < pOption->numOfTotalReplicas; ++i) {
  }

  return 0;
}

int32_t mmReadFile(const char *path, SMnodeOpt *pOption) {
  int32_t   code = -1;
  TdFilePtr pFile = NULL;
  char     *pData = NULL;
  SJson    *pJson = NULL;
  char      file[PATH_MAX] = {0};

  int32_t nBytes = snprintf(file, sizeof(file), "%s%smnode.json", path, TD_DIRSEP);
  if (nBytes <= 0 || nBytes >= sizeof(file)) {
    code = TSDB_CODE_OUT_OF_BUFFER;
    goto _OVER;
  }

  if (taosStatFile(file, NULL, NULL, NULL) < 0) {
    dInfo("mnode file:%s not exist, reason:%s", file, tstrerror(terrno));
    return 0;
  }

  pFile = taosOpenFile(file, TD_FILE_READ);
  if (pFile == NULL) {
    code = terrno;
    dError("failed to open mnode file:%s since %s", file, tstrerror(code));
    goto _OVER;
  }

  int64_t size = 0;
  code = taosFStatFile(pFile, &size, NULL);
  if (code != 0) {
    dError("failed to fstat mnode file:%s since %s", file, tstrerror(code));
    goto _OVER;
  }

  pData = taosMemoryMalloc(size + 1);
  if (pData == NULL) {
    code = terrno;
    goto _OVER;
  }

  if (taosReadFile(pFile, pData, size) != size) {
    code = terrno;
    dError("failed to read mnode file:%s since %s", file, tstrerror(code));
    goto _OVER;
  }

  pData[size] = '\0';

  pJson = tjsonParse(pData);
  if (pJson == NULL) {
    code = TSDB_CODE_INVALID_JSON_FORMAT;
    goto _OVER;
  }

  if ((code = mmDecodeOption(pJson, pOption)) < 0) {
    goto _OVER;
  }

  code = 0;
  dInfo("succceed to read mnode file %s", file);

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

  if (code != 0) {
    dError("failed to read mnode file:%s since %s", file, tstrerror(code));
  }
  return code;
}

static int32_t mmEncodeOption(SJson *pJson, const SMnodeOpt *pOption) {
  int32_t code = 0;
  if (pOption->deploy && pOption->numOfTotalReplicas > 0) {
    if ((code = tjsonAddDoubleToObject(pJson, "selfIndex", pOption->selfIndex)) < 0) return code;

    SJson *replicas = tjsonCreateArray();
    if (replicas == NULL) {
      return terrno;
    }
    if ((code = tjsonAddItemToObject(pJson, "replicas", replicas)) < 0) return code;

    for (int32_t i = 0; i < pOption->numOfTotalReplicas; ++i) {
      SJson *replica = tjsonCreateObject();
      if (replica == NULL) {
        return terrno;
      }

      const SReplica *pReplica = pOption->replicas + i;
      if ((code = tjsonAddDoubleToObject(replica, "id", pReplica->id)) < 0) return code;
      if ((code = tjsonAddStringToObject(replica, "fqdn", pReplica->fqdn)) < 0) return code;
      if ((code = tjsonAddDoubleToObject(replica, "port", pReplica->port)) < 0) return code;
      if ((code = tjsonAddDoubleToObject(replica, "role", pOption->nodeRoles[i])) < 0) return code;
      if ((code = tjsonAddItemToArray(replicas, replica)) < 0) return code;
    }
  }

  if ((code = tjsonAddDoubleToObject(pJson, "lastIndex", pOption->lastIndex)) < 0) return code;

  if ((code = tjsonAddDoubleToObject(pJson, "deployed", pOption->deploy)) < 0) return code;

  return code;
}

int32_t mmWriteFile(const char *path, const SMnodeOpt *pOption) {
  int32_t   code = -1;
  char     *buffer = NULL;
  SJson    *pJson = NULL;
  TdFilePtr pFile = NULL;
  char      file[PATH_MAX] = {0};
  char      realfile[PATH_MAX] = {0};

  int32_t nBytes = snprintf(file, sizeof(file), "%s%smnode.json.bak", path, TD_DIRSEP);
  if (nBytes <= 0 || nBytes >= sizeof(file)) {
    code = TSDB_CODE_OUT_OF_BUFFER;
    goto _OVER;
  }

  nBytes = snprintf(realfile, sizeof(realfile), "%s%smnode.json", path, TD_DIRSEP);
  if (nBytes <= 0 || nBytes >= sizeof(realfile)) {
    code = TSDB_CODE_OUT_OF_BUFFER;
    goto _OVER;
  }

  // terrno = TSDB_CODE_OUT_OF_MEMORY;
  pJson = tjsonCreateObject();
  if (pJson == NULL) {
    code = terrno;
    goto _OVER;
  }

  TAOS_CHECK_GOTO(mmEncodeOption(pJson, pOption), NULL, _OVER);

  buffer = tjsonToString(pJson);
  if (buffer == NULL) {
    code = TSDB_CODE_INVALID_JSON_FORMAT;
    goto _OVER;
  }

  pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
  if (pFile == NULL) {
    code = terrno;
    goto _OVER;
  }

  int32_t len = strlen(buffer);
  if (taosWriteFile(pFile, buffer, len) <= 0) {
    code = terrno;
    goto _OVER;
  }
  if (taosFsyncFile(pFile) < 0) {
    code = terrno;
    goto _OVER;
  }

  if (taosCloseFile(&pFile) < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    goto _OVER;
  }
  TAOS_CHECK_GOTO(taosRenameFile(file, realfile), NULL, _OVER);

  dInfo("succeed to write mnode file:%s, deloyed:%d", realfile, pOption->deploy);

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

  if (code != 0) {
    dError("failed to write mnode file:%s since %s, deloyed:%d", realfile, tstrerror(code), pOption->deploy);
  }
  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	#define _DEFAULT_SOURCE
17	#include "mmInt.h"
18	#include "tjson.h"
19
20	static int32_t mmDecodeOption(SJson pJson, SMnodeOpt pOption) {	2✔
21	int32_t code = 0;	2✔
22
23	tjsonGetInt32ValueFromDouble(pJson, "deployed", pOption->deploy, code);	2✔
24	if (code < 0) return code;	2!
25	tjsonGetInt32ValueFromDouble(pJson, "selfIndex", pOption->selfIndex, code);	2✔
26	if (code < 0) return code;	2!
27	tjsonGetInt32ValueFromDouble(pJson, "lastIndex", pOption->lastIndex, code);	2✔
28	if (code < 0) return code;	2!
29
30	SJson *replicas = tjsonGetObjectItem(pJson, "replicas");	2✔
31	if (replicas == NULL) return 0;	2!
UNCOV 32	pOption->numOfTotalReplicas = tjsonGetArraySize(replicas);	×
33
UNCOV 34	pOption->numOfReplicas = 0;	×
35
UNCOV 36	for (int32_t i = 0; i < pOption->numOfTotalReplicas; ++i) {	×
UNCOV 37	SJson *replica = tjsonGetArrayItem(replicas, i);	×
UNCOV 38	if (replica == NULL) return TSDB_CODE_INVALID_JSON_FORMAT;	×
39
UNCOV 40	SReplica *pReplica = pOption->replicas + i;	×
UNCOV 41	tjsonGetInt32ValueFromDouble(replica, "id", pReplica->id, code);	×
UNCOV 42	if (code < 0) return code;	×
UNCOV 43	code = tjsonGetStringValue(replica, "fqdn", pReplica->fqdn);	×
UNCOV 44	if (code < 0) return code;	×
UNCOV 45	tjsonGetUInt16ValueFromDouble(replica, "port", pReplica->port, code);	×
UNCOV 46	if (code < 0) return code;	×
UNCOV 47	tjsonGetInt32ValueFromDouble(replica, "role", pOption->nodeRoles[i], code);	×
UNCOV 48	if (code < 0) return code;	×
UNCOV 49	if (pOption->nodeRoles[i] == TAOS_SYNC_ROLE_VOTER) {	×
UNCOV 50	pOption->numOfReplicas++;	×
51	}
52	}
53
UNCOV 54	for (int32_t i = 0; i < pOption->numOfTotalReplicas; ++i) {	×
55	}
56
UNCOV 57	return 0;	×
58	}
59
60	int32_t mmReadFile(const char path, SMnodeOpt pOption) {	56✔
61	int32_t code = -1;	56✔
62	TdFilePtr pFile = NULL;	56✔
63	char *pData = NULL;	56✔
64	SJson *pJson = NULL;	56✔
65	char file[PATH_MAX] = {0};	56✔
66
67	int32_t nBytes = snprintf(file, sizeof(file), "%s%smnode.json", path, TD_DIRSEP);	56✔
68	if (nBytes <= 0 \|\| nBytes >= sizeof(file)) {	56!
69	code = TSDB_CODE_OUT_OF_BUFFER;	×
70	goto _OVER;	×
71	}
72
73	if (taosStatFile(file, NULL, NULL, NULL) < 0) {	56✔
74	dInfo("mnode file:%s not exist, reason:%s", file, tstrerror(terrno));	54!
75	return 0;	54✔
76	}
77
78	pFile = taosOpenFile(file, TD_FILE_READ);	2✔
79	if (pFile == NULL) {	2!
80	code = terrno;	×
81	dError("failed to open mnode file:%s since %s", file, tstrerror(code));	×
82	goto _OVER;	×
83	}
84
85	int64_t size = 0;	2✔
86	code = taosFStatFile(pFile, &size, NULL);	2✔
87	if (code != 0) {	2!
88	dError("failed to fstat mnode file:%s since %s", file, tstrerror(code));	×
89	goto _OVER;	×
90	}
91
92	pData = taosMemoryMalloc(size + 1);	2!
93	if (pData == NULL) {	2!
94	code = terrno;	×
95	goto _OVER;	×
96	}
97
98	if (taosReadFile(pFile, pData, size) != size) {	2!
99	code = terrno;	×
100	dError("failed to read mnode file:%s since %s", file, tstrerror(code));	×
101	goto _OVER;	×
102	}
103
104	pData[size] = '\0';	2✔
105
106	pJson = tjsonParse(pData);	2✔
107	if (pJson == NULL) {	2!
108	code = TSDB_CODE_INVALID_JSON_FORMAT;	×
109	goto _OVER;	×
110	}
111
112	if ((code = mmDecodeOption(pJson, pOption)) < 0) {	2!
113	goto _OVER;	×
114	}
115
116	code = 0;	2✔
117	dInfo("succceed to read mnode file %s", file);	2!
118
119	_OVER:	×
120	if (pData != NULL) taosMemoryFree(pData);	2!
121	if (pJson != NULL) cJSON_Delete(pJson);	2!
122	if (pFile != NULL) taosCloseFile(&pFile);	2!
123
124	if (code != 0) {	2!
125	dError("failed to read mnode file:%s since %s", file, tstrerror(code));	×
126	}
127	return code;	2✔
128	}
129
130	static int32_t mmEncodeOption(SJson pJson, const SMnodeOpt pOption) {	12✔
131	int32_t code = 0;	12✔
132	if (pOption->deploy && pOption->numOfTotalReplicas > 0) {	12!
UNCOV 133	if ((code = tjsonAddDoubleToObject(pJson, "selfIndex", pOption->selfIndex)) < 0) return code;	×
134
UNCOV 135	SJson *replicas = tjsonCreateArray();	×
UNCOV 136	if (replicas == NULL) {	×
137	return terrno;	×
138	}
UNCOV 139	if ((code = tjsonAddItemToObject(pJson, "replicas", replicas)) < 0) return code;	×
140
UNCOV 141	for (int32_t i = 0; i < pOption->numOfTotalReplicas; ++i) {	×
UNCOV 142	SJson *replica = tjsonCreateObject();	×
UNCOV 143	if (replica == NULL) {	×
144	return terrno;	×
145	}
146
UNCOV 147	const SReplica *pReplica = pOption->replicas + i;	×
UNCOV 148	if ((code = tjsonAddDoubleToObject(replica, "id", pReplica->id)) < 0) return code;	×
UNCOV 149	if ((code = tjsonAddStringToObject(replica, "fqdn", pReplica->fqdn)) < 0) return code;	×
UNCOV 150	if ((code = tjsonAddDoubleToObject(replica, "port", pReplica->port)) < 0) return code;	×
UNCOV 151	if ((code = tjsonAddDoubleToObject(replica, "role", pOption->nodeRoles[i])) < 0) return code;	×
UNCOV 152	if ((code = tjsonAddItemToArray(replicas, replica)) < 0) return code;	×
153	}
154	}
155
156	if ((code = tjsonAddDoubleToObject(pJson, "lastIndex", pOption->lastIndex)) < 0) return code;	12!
157
158	if ((code = tjsonAddDoubleToObject(pJson, "deployed", pOption->deploy)) < 0) return code;	12!
159
160	return code;	12✔
161	}
162
163	int32_t mmWriteFile(const char path, const SMnodeOpt pOption) {	12✔
164	int32_t code = -1;	12✔
165	char *buffer = NULL;	12✔
166	SJson *pJson = NULL;	12✔
167	TdFilePtr pFile = NULL;	12✔
168	char file[PATH_MAX] = {0};	12✔
169	char realfile[PATH_MAX] = {0};	12✔
170
171	int32_t nBytes = snprintf(file, sizeof(file), "%s%smnode.json.bak", path, TD_DIRSEP);	12✔
172	if (nBytes <= 0 \|\| nBytes >= sizeof(file)) {	12!
173	code = TSDB_CODE_OUT_OF_BUFFER;	×
174	goto _OVER;	×
175	}
176
177	nBytes = snprintf(realfile, sizeof(realfile), "%s%smnode.json", path, TD_DIRSEP);	12✔
178	if (nBytes <= 0 \|\| nBytes >= sizeof(realfile)) {	12!
179	code = TSDB_CODE_OUT_OF_BUFFER;	×
180	goto _OVER;	×
181	}
182
183	// terrno = TSDB_CODE_OUT_OF_MEMORY;
184	pJson = tjsonCreateObject();	12✔
185	if (pJson == NULL) {	12!
186	code = terrno;	×
187	goto _OVER;	×
188	}
189
190	TAOS_CHECK_GOTO(mmEncodeOption(pJson, pOption), NULL, _OVER);	12!
191
192	buffer = tjsonToString(pJson);	12✔
193	if (buffer == NULL) {	12!
194	code = TSDB_CODE_INVALID_JSON_FORMAT;	×
195	goto _OVER;	×
196	}
197
198	pFile = taosOpenFile(file, TD_FILE_CREATE \| TD_FILE_WRITE \| TD_FILE_TRUNC \| TD_FILE_WRITE_THROUGH);	12✔
199	if (pFile == NULL) {	12!
200	code = terrno;	×
201	goto _OVER;	×
202	}
203
204	int32_t len = strlen(buffer);	12✔
205	if (taosWriteFile(pFile, buffer, len) <= 0) {	12!
206	code = terrno;	×
207	goto _OVER;	×
208	}
209	if (taosFsyncFile(pFile) < 0) {	12!
210	code = terrno;	×
211	goto _OVER;	×
212	}
213
214	if (taosCloseFile(&pFile) < 0) {	12!
215	code = TAOS_SYSTEM_ERROR(errno);	×
216	goto _OVER;	×
217	}
218	TAOS_CHECK_GOTO(taosRenameFile(file, realfile), NULL, _OVER);	12!
219
220	dInfo("succeed to write mnode file:%s, deloyed:%d", realfile, pOption->deploy);	12!
221
222	_OVER:	×
223	if (pJson != NULL) tjsonDelete(pJson);	12!
224	if (buffer != NULL) taosMemoryFree(buffer);	12!
225	if (pFile != NULL) taosCloseFile(&pFile);	12!
226
227	if (code != 0) {	12!
228	dError("failed to write mnode file:%s since %s, deloyed:%d", realfile, tstrerror(code), pOption->deploy);	×
229	}
230	return code;	12✔
231	}

taosdata / TDengine / #3562

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