xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request #26790 from taosdata/enh/TD-30988-3.0 

fix: refactor return code
This commit is contained in:
Hongze Cheng 2024-07-26 20:33:50 +08:00 committed by GitHub
parent 97dc483680 87e3c17ada
commit 3460a202fc
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
29 changed files with 855 additions and 776 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

44
source/common/src/trow.c
View File

@ -138,7 +138,7 @@ int32_t tdGetBitmapValTypeII(const void *pBitmap, int16_t colIdx, TDRowValT *pVa
int32_t tdGetBitmapValType(const void *pBitmap, int16_t colIdx, TDRowValT *pValType, int8_t bitmapMode) {
switch (bitmapMode) {
case 0:
tdGetBitmapValTypeII(pBitmap, colIdx, pValType);
(void)tdGetBitmapValTypeII(pBitmap, colIdx, pValType);
break;
#if 0
case -1:
@ -354,8 +354,8 @@ bool tdSKvRowGetVal(STSRow *pRow, col_id_t colId, col_id_t colIdx, SCellVal *pVa
}
void *pBitmap = tdGetBitmapAddrKv(pRow, tdRowGetNCols(pRow));
tdGetKvRowValOfCol(pVal, pRow, pBitmap, pColIdx->offset,
POINTER_DISTANCE(pColIdx, TD_ROW_COL_IDX(pRow)) / sizeof(SKvRowIdx));
(void)tdGetKvRowValOfCol(pVal, pRow, pBitmap, pColIdx->offset,
POINTER_DISTANCE(pColIdx, TD_ROW_COL_IDX(pRow)) / sizeof(SKvRowIdx));
return true;
}
@ -366,7 +366,7 @@ bool tdSTpRowGetVal(STSRow *pRow, col_id_t colId, col_type_t colType, int32_t fl
return true;
}
void *pBitmap = tdGetBitmapAddrTp(pRow, flen);
tdGetTpRowValOfCol(pVal, pRow, pBitmap, colType, offset, colIdx);
(void)tdGetTpRowValOfCol(pVal, pRow, pBitmap, colType, offset, colIdx);
return true;
}
@ -385,9 +385,9 @@ bool tdSTSRowIterNext(STSRowIter *pIter, SCellVal *pVal) {
}
if (TD_IS_TP_ROW(pIter->pRow)) {
tdSTSRowIterGetTpVal(pIter, pCol->type, pCol->offset, pVal);
(void)tdSTSRowIterGetTpVal(pIter, pCol->type, pCol->offset, pVal);
} else if (TD_IS_KV_ROW(pIter->pRow)) {
tdSTSRowIterGetKvVal(pIter, pCol->colId, &pIter->kvIdx, pVal);
(void)tdSTSRowIterGetKvVal(pIter, pCol->colId, &pIter->kvIdx, pVal);
} else {
ASSERT(0);
}
@ -486,8 +486,8 @@ int32_t tdSTSRowNew(SArray *pArray, STSchema *pTSchema, STSRow **ppRow, int8_t r
SRowBuilder rb = {.rowType = rowType};
tdSRowInit(&rb, pTSchema->version);
tdSRowSetInfo(&rb, pTSchema->numOfCols, nBound, pTSchema->flen);
tdSRowResetBuf(&rb, *ppRow);
(void)tdSRowSetInfo(&rb, pTSchema->numOfCols, nBound, pTSchema->flen);
(void)tdSRowResetBuf(&rb, *ppRow);
int32_t iBound = 0;
iColVal = 0;
@ -506,7 +506,7 @@ int32_t tdSTSRowNew(SArray *pArray, STSchema *pTSchema, STSRow **ppRow, int8_t r
} else if (IS_VAR_DATA_TYPE(pTColumn->type)) {
varDataSetLen(varBuf, pColVal->value.nData);
if (pColVal->value.nData != 0) {
memcpy(varDataVal(varBuf), pColVal->value.pData, pColVal->value.nData);
(void)memcpy(varDataVal(varBuf), pColVal->value.pData, pColVal->value.nData);
}
val = varBuf;
++iBound;
@ -578,7 +578,7 @@ bool tdSTSRowGetVal(STSRowIter *pIter, col_id_t colId, col_type_t colType, SCell
#ifdef TD_SUPPORT_BITMAP
colIdx = POINTER_DISTANCE(pCol, pSchema->columns) / sizeof(STColumn);
#endif
tdGetTpRowValOfCol(pVal, pRow, pIter->pBitmap, pCol->type, pCol->offset, colIdx - 1);
(void)tdGetTpRowValOfCol(pVal, pRow, pIter->pBitmap, pCol->type, pCol->offset, colIdx - 1);
} else if (TD_IS_KV_ROW(pRow)) {
SKvRowIdx *pIdx = (SKvRowIdx *)taosbsearch(&colId, TD_ROW_COL_IDX(pRow), tdRowGetNCols(pRow), sizeof(SKvRowIdx),
compareKvRowColId, TD_EQ);
@ -587,7 +587,7 @@ bool tdSTSRowGetVal(STSRowIter *pIter, col_id_t colId, col_type_t colType, SCell
colIdx = POINTER_DISTANCE(pIdx, TD_ROW_COL_IDX(pRow)) / sizeof(SKvRowIdx);
}
#endif
tdGetKvRowValOfCol(pVal, pRow, pIter->pBitmap, pIdx ? pIdx->offset : -1, colIdx);
(void)tdGetKvRowValOfCol(pVal, pRow, pIter->pBitmap, pIdx ? pIdx->offset : -1, colIdx);
} else {
if (COL_REACH_END(colId, pIter->maxColId)) return false;
pVal->valType = TD_VTYPE_NONE;
@ -689,9 +689,9 @@ int32_t tdAppendColValToRow(SRowBuilder *pBuilder, col_id_t colId, int8_t colTyp
}
if (TD_IS_TP_ROW(pRow)) {
tdAppendColValToTpRow(pBuilder, valType, val, isCopyVarData, colType, colIdx, offset);
(void)tdAppendColValToTpRow(pBuilder, valType, val, isCopyVarData, colType, colIdx, offset);
} else {
tdAppendColValToKvRow(pBuilder, valType, val, isCopyVarData, colType, colIdx, offset, colId);
(void)tdAppendColValToKvRow(pBuilder, valType, val, isCopyVarData, colType, colIdx, offset, colId);
}
TAOS_RETURN(TSDB_CODE_SUCCESS);
}
@ -718,11 +718,11 @@ int32_t tdAppendColValToKvRow(SRowBuilder *pBuilder, TDRowValT valType, const vo
char *ptr = (char *)POINTER_SHIFT(row, TD_ROW_LEN(row));
if (IS_VAR_DATA_TYPE(colType)) {
if (isCopyVarData) {
memcpy(ptr, val, varDataTLen(val));
(void)memcpy(ptr, val, varDataTLen(val));
}
TD_ROW_LEN(row) += varDataTLen(val);
} else {
memcpy(ptr, val, TYPE_BYTES[colType]);
(void)memcpy(ptr, val, TYPE_BYTES[colType]);
TD_ROW_LEN(row) += TYPE_BYTES[colType];
}
}
@ -752,11 +752,11 @@ int32_t tdAppendColValToTpRow(SRowBuilder *pBuilder, TDRowValT valType, const vo
// ts key stored in STSRow.ts
*(VarDataOffsetT *)POINTER_SHIFT(TD_ROW_DATA(row), offset) = TD_ROW_LEN(row);
if (isCopyVarData) {
memcpy(POINTER_SHIFT(row, TD_ROW_LEN(row)), val, varDataTLen(val));
(void)memcpy(POINTER_SHIFT(row, TD_ROW_LEN(row)), val, varDataTLen(val));
}
TD_ROW_LEN(row) += varDataTLen(val);
} else {
memcpy(POINTER_SHIFT(TD_ROW_DATA(row), offset), val, TYPE_BYTES[colType]);
(void)memcpy(POINTER_SHIFT(TD_ROW_DATA(row), offset), val, TYPE_BYTES[colType]);
}
}
@ -782,7 +782,7 @@ int32_t tdSRowResetBuf(SRowBuilder *pBuilder, void *pBuf) {
case TD_ROW_TP:
#ifdef TD_SUPPORT_BITMAP
pBuilder->pBitmap = tdGetBitmapAddrTp(pBuilder->pBuf, pBuilder->flen);
memset(pBuilder->pBitmap, TD_VTYPE_NONE_BYTE_II, pBuilder->nBitmaps);
(void)memset(pBuilder->pBitmap, TD_VTYPE_NONE_BYTE_II, pBuilder->nBitmaps);
#endif
// the primary TS key is stored separatedly
len = TD_ROW_HEAD_LEN + pBuilder->flen + pBuilder->nBitmaps;
@ -792,7 +792,7 @@ int32_t tdSRowResetBuf(SRowBuilder *pBuilder, void *pBuf) {
case TD_ROW_KV:
#ifdef TD_SUPPORT_BITMAP
pBuilder->pBitmap = tdGetBitmapAddrKv(pBuilder->pBuf, pBuilder->nBoundCols);
memset(pBuilder->pBitmap, TD_VTYPE_NONE_BYTE_II, pBuilder->nBoundBitmaps);
(void)memset(pBuilder->pBitmap, TD_VTYPE_NONE_BYTE_II, pBuilder->nBoundBitmaps);
#endif
len = TD_ROW_HEAD_LEN + TD_ROW_NCOLS_LEN + (pBuilder->nBoundCols - 1) * sizeof(SKvRowIdx) +
pBuilder->nBoundBitmaps; // add
@ -937,9 +937,9 @@ void tTSRowGetVal(STSRow *pRow, STSchema *pTSchema, int16_t iCol, SColVal *pColV
ASSERT((pTColumn->colId == PRIMARYKEY_TIMESTAMP_COL_ID) || (iCol > 0));
if (TD_IS_TP_ROW(pRow)) {
tdSTpRowGetVal(pRow, pTColumn->colId, pTColumn->type, pTSchema->flen, pTColumn->offset, iCol - 1, &cv);
(void)tdSTpRowGetVal(pRow, pTColumn->colId, pTColumn->type, pTSchema->flen, pTColumn->offset, iCol - 1, &cv);
} else if (TD_IS_KV_ROW(pRow)) {
tdSKvRowGetVal(pRow, pTColumn->colId, iCol - 1, &cv);
(void)tdSKvRowGetVal(pRow, pTColumn->colId, iCol - 1, &cv);
} else {
ASSERT(0);
}
@ -957,7 +957,7 @@ void tTSRowGetVal(STSRow *pRow, STSchema *pTSchema, int16_t iCol, SColVal *pColV
pColVal->value.nData = varDataLen(cv.val);
pColVal->value.pData = varDataVal(cv.val);
} else {
memcpy(&pColVal->value.val, cv.val, tDataTypes[pTColumn->type].bytes);
(void)memcpy(&pColVal->value.val, cv.val, tDataTypes[pTColumn->type].bytes);
}
}
}

140
source/dnode/mnode/impl/src/mndUser.c
View File

@ -142,12 +142,12 @@ static int32_t ipWhiteMgtInit() {
TAOS_RETURN(TSDB_CODE_OUT_OF_MEMORY);
}
ipWhiteMgt.ver = 0;
taosThreadRwlockInit(&ipWhiteMgt.rw, NULL);
(void)taosThreadRwlockInit(&ipWhiteMgt.rw, NULL);
TAOS_RETURN(0);
}
void ipWhiteMgtCleanup() {
destroyIpWhiteTab(ipWhiteMgt.pIpWhiteTab);
taosThreadRwlockDestroy(&ipWhiteMgt.rw);
(void)taosThreadRwlockDestroy(&ipWhiteMgt.rw);
}
int32_t ipWhiteMgtUpdate(SMnode *pMnode, char *user, SIpWhiteList *pNew) {
@ -155,7 +155,7 @@ int32_t ipWhiteMgtUpdate(SMnode *pMnode, char *user, SIpWhiteList *pNew) {
int32_t lino = 0;
bool update = true;
SArray *fqdns = NULL;
taosThreadRwlockWrlock(&ipWhiteMgt.rw);
(void)taosThreadRwlockWrlock(&ipWhiteMgt.rw);
SIpWhiteList **ppList = taosHashGet(ipWhiteMgt.pIpWhiteTab, user, strlen(user));
if (ppList == NULL || *ppList == NULL) {
@ -212,7 +212,7 @@ int32_t ipWhiteMgtUpdate(SMnode *pMnode, char *user, SIpWhiteList *pNew) {
if (update) ipWhiteMgt.ver++;
_OVER:
taosThreadRwlockUnlock(&ipWhiteMgt.rw);
(void)taosThreadRwlockUnlock(&ipWhiteMgt.rw);
taosArrayDestroyP(fqdns, (FDelete)taosMemoryFree);
if (code < 0) {
mError("failed to update ip white list for user: %s at line %d since %s", user, lino, tstrerror(code));
@ -221,17 +221,17 @@ _OVER:
}
int32_t ipWhiteMgtRemove(char *user) {
bool update = true;
taosThreadRwlockWrlock(&ipWhiteMgt.rw);
(void)taosThreadRwlockWrlock(&ipWhiteMgt.rw);
SIpWhiteList **ppList = taosHashGet(ipWhiteMgt.pIpWhiteTab, user, strlen(user));
if (ppList == NULL || *ppList == NULL) {
update = false;
} else {
taosMemoryFree(*ppList);
taosHashRemove(ipWhiteMgt.pIpWhiteTab, user, strlen(user));
(void)taosHashRemove(ipWhiteMgt.pIpWhiteTab, user, strlen(user));
}
if (update) ipWhiteMgt.ver++;
taosThreadRwlockUnlock(&ipWhiteMgt.rw);
(void)taosThreadRwlockUnlock(&ipWhiteMgt.rw);
return 0;
}
@ -245,7 +245,7 @@ bool isRangeInWhiteList(SIpWhiteList *pList, SIpV4Range *range) {
}
#if 0
int32_t ipWhiteUpdateForAllUser(SIpWhiteList *pList) {
taosThreadRwlockWrlock(&ipWhiteMgt.rw);
(void)taosThreadRwlockWrlock(&ipWhiteMgt.rw);
SHashObj *pIpWhiteTab = taosHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), 1, HASH_ENTRY_LOCK);
void *pIter = taosHashIterate(ipWhiteMgt.pIpWhiteTab, NULL);
@ -267,8 +267,8 @@ int32_t ipWhiteUpdateForAllUser(SIpWhiteList *pList) {
if (clone->num != 0) {
int32_t sz = clone->num + p->num;
val = taosMemoryCalloc(1, sizeof(SIpWhiteList) + sz * sizeof(SIpV4Range));
memcpy(val->pIpRange, p->pIpRange, sizeof(SIpV4Range) * p->num);
memcpy(((char *)val->pIpRange) + sizeof(SIpV4Range) * p->num, (char *)clone->pIpRange,
(void)memcpy(val->pIpRange, p->pIpRange, sizeof(SIpV4Range) * p->num);
(void)memcpy(((char *)val->pIpRange) + sizeof(SIpV4Range) * p->num, (char *)clone->pIpRange,
sizeof(SIpV4Range) * clone->num);
} else {
@ -285,7 +285,7 @@ int32_t ipWhiteUpdateForAllUser(SIpWhiteList *pList) {
ipWhiteMgt.pIpWhiteTab = pIpWhiteTab;
ipWhiteMgt.ver++;
taosThreadRwlockUnlock(&ipWhiteMgt.rw);
(void)taosThreadRwlockUnlock(&ipWhiteMgt.rw);
return 0;
}
#endif
@ -305,29 +305,29 @@ static int32_t ipWhiteMgtUpdateAll(SMnode *pMnode) {
#if 0
void ipWhiteMgtUpdate2(SMnode *pMnode) {
taosThreadRwlockWrlock(&ipWhiteMgt.rw);
(void)taosThreadRwlockWrlock(&ipWhiteMgt.rw);
ipWhiteMgtUpdateAll(pMnode);
taosThreadRwlockUnlock(&ipWhiteMgt.rw);
(void)taosThreadRwlockUnlock(&ipWhiteMgt.rw);
}
#endif
int64_t mndGetIpWhiteVer(SMnode *pMnode) {
int64_t ver = 0;
int32_t code = 0;
taosThreadRwlockWrlock(&ipWhiteMgt.rw);
(void)taosThreadRwlockWrlock(&ipWhiteMgt.rw);
if (ipWhiteMgt.ver == 0) {
// get user and dnode ip white list
if ((code = ipWhiteMgtUpdateAll(pMnode)) != 0) {
taosThreadRwlockUnlock(&ipWhiteMgt.rw);
(void)taosThreadRwlockUnlock(&ipWhiteMgt.rw);
mError("%s failed to update ip white list since %s", __func__, tstrerror(code));
return ver;
}
ipWhiteMgt.ver = taosGetTimestampMs();
}
ver = ipWhiteMgt.ver;
taosThreadRwlockUnlock(&ipWhiteMgt.rw);
(void)taosThreadRwlockUnlock(&ipWhiteMgt.rw);
if (mndEnableIpWhiteList(pMnode) == 0 || tsEnableWhiteList == false) {
ver = 0;
@ -357,7 +357,7 @@ int32_t mndUpdateIpWhiteImpl(SHashObj *pIpWhiteTab, char *user, char *fqdn, int8
if (pNewList == NULL) {
TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _OVER);
}
memcpy(pNewList->pIpRange, &range, sizeof(SIpV4Range));
(void)memcpy(pNewList->pIpRange, &range, sizeof(SIpV4Range));
pNewList->num = 1;
if ((code = taosHashPut(pIpWhiteTab, user, strlen(user), &pNewList, sizeof(void *))) != 0) {
@ -372,7 +372,7 @@ int32_t mndUpdateIpWhiteImpl(SHashObj *pIpWhiteTab, char *user, char *fqdn, int8
if (pNewList == NULL) {
TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _OVER);
}
memcpy(pNewList->pIpRange, pList->pIpRange, sizeof(SIpV4Range) * (pList->num));
(void)memcpy(pNewList->pIpRange, pList->pIpRange, sizeof(SIpV4Range) * (pList->num));
pNewList->pIpRange[pList->num].ip = range.ip;
pNewList->pIpRange[pList->num].mask = range.mask;
@ -390,7 +390,7 @@ int32_t mndUpdateIpWhiteImpl(SHashObj *pIpWhiteTab, char *user, char *fqdn, int8
if (pList != NULL) {
if (isRangeInWhiteList(pList, &range)) {
if (pList->num == 1) {
taosHashRemove(pIpWhiteTab, user, strlen(user));
(void)taosHashRemove(pIpWhiteTab, user, strlen(user));
taosMemoryFree(pList);
} else {
int32_t idx = 0;
@ -433,14 +433,14 @@ _OVER:
int32_t mndRefreshUserIpWhiteList(SMnode *pMnode) {
int32_t code = 0;
taosThreadRwlockWrlock(&ipWhiteMgt.rw);
(void)taosThreadRwlockWrlock(&ipWhiteMgt.rw);
if ((code = ipWhiteMgtUpdateAll(pMnode)) != 0) {
taosThreadRwlockUnlock(&ipWhiteMgt.rw);
(void)taosThreadRwlockUnlock(&ipWhiteMgt.rw);
TAOS_RETURN(code);
}
ipWhiteMgt.ver = taosGetTimestampMs();
taosThreadRwlockUnlock(&ipWhiteMgt.rw);
(void)taosThreadRwlockUnlock(&ipWhiteMgt.rw);
TAOS_RETURN(code);
}
@ -451,7 +451,7 @@ int32_t mndUpdateIpWhiteForAllUser(SMnode *pMnode, char *user, char *fqdn, int8_
bool update = false;
if (lock) {
taosThreadRwlockWrlock(&ipWhiteMgt.rw);
(void)taosThreadRwlockWrlock(&ipWhiteMgt.rw);
if (ipWhiteMgt.ver == 0) {
TAOS_CHECK_GOTO(ipWhiteMgtUpdateAll(pMnode), &lino, _OVER);
ipWhiteMgt.ver = taosGetTimestampMs();
@ -470,7 +470,7 @@ int32_t mndUpdateIpWhiteForAllUser(SMnode *pMnode, char *user, char *fqdn, int8_
if (keyDup == NULL) {
TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _OVER);
}
memcpy(keyDup, key, klen);
(void)memcpy(keyDup, key, klen);
bool upd = false;
code = mndUpdateIpWhiteImpl(ipWhiteMgt.pIpWhiteTab, keyDup, fqdn, type, &upd);
update |= upd;
@ -485,7 +485,7 @@ int32_t mndUpdateIpWhiteForAllUser(SMnode *pMnode, char *user, char *fqdn, int8_
_OVER:
if (update) ipWhiteMgt.ver++;
if (lock) taosThreadRwlockUnlock(&ipWhiteMgt.rw);
if (lock) (void)taosThreadRwlockUnlock(&ipWhiteMgt.rw);
if (code < 0) {
mError("failed to update ip-white-list for user: %s, fqdn: %s at line %d since %s", user, fqdn, lino,
tstrerror(code));
@ -496,13 +496,13 @@ _OVER:
static int64_t ipWhiteMgtFillMsg(SUpdateIpWhite *pUpdate) {
int64_t ver = 0;
taosThreadRwlockWrlock(&ipWhiteMgt.rw);
(void)taosThreadRwlockWrlock(&ipWhiteMgt.rw);
ver = ipWhiteMgt.ver;
int32_t num = taosHashGetSize(ipWhiteMgt.pIpWhiteTab);
pUpdate->pUserIpWhite = taosMemoryCalloc(1, num * sizeof(SUpdateUserIpWhite));
if (pUpdate->pUserIpWhite == NULL) {
taosThreadRwlockUnlock(&ipWhiteMgt.rw);
(void)taosThreadRwlockUnlock(&ipWhiteMgt.rw);
TAOS_RETURN(TSDB_CODE_OUT_OF_MEMORY);
}
@ -516,14 +516,14 @@ static int64_t ipWhiteMgtFillMsg(SUpdateIpWhite *pUpdate) {
char *key = taosHashGetKey(pIter, &klen);
if (list->num != 0) {
pUser->ver = ver;
memcpy(pUser->user, key, klen);
(void)memcpy(pUser->user, key, klen);
pUser->numOfRange = list->num;
pUser->pIpRanges = taosMemoryCalloc(1, list->num * sizeof(SIpV4Range));
if (pUser->pIpRanges == NULL) {
taosThreadRwlockUnlock(&ipWhiteMgt.rw);
(void)taosThreadRwlockUnlock(&ipWhiteMgt.rw);
TAOS_RETURN(TSDB_CODE_OUT_OF_MEMORY);
}
memcpy(pUser->pIpRanges, list->pIpRange, list->num * sizeof(SIpV4Range));
(void)memcpy(pUser->pIpRanges, list->pIpRange, list->num * sizeof(SIpV4Range));
i++;
}
pIter = taosHashIterate(ipWhiteMgt.pIpWhiteTab, pIter);
@ -531,7 +531,7 @@ static int64_t ipWhiteMgtFillMsg(SUpdateIpWhite *pUpdate) {
pUpdate->numOfUser = i;
pUpdate->ver = ver;
taosThreadRwlockUnlock(&ipWhiteMgt.rw);
(void)taosThreadRwlockUnlock(&ipWhiteMgt.rw);
TAOS_RETURN(0);
}
@ -684,7 +684,7 @@ static void ipRangeToStr(SIpV4Range *range, char *buf) {
uv_inet_ntop(AF_INET, &addr, buf, 32);
if (range->mask != 32) {
sprintf(buf + strlen(buf), "/%d", range->mask);
(void)sprintf(buf + strlen(buf), "/%d", range->mask);
}
return;
}
@ -1314,7 +1314,7 @@ static SSdbRow *mndUserActionDecode(SSdbRaw *pRaw) {
if (key == NULL) {
TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _OVER);
}
memset(key, 0, keyLen);
(void)memset(key, 0, keyLen);
SDB_GET_BINARY(pRaw, dataPos, key, keyLen, _OVER);
int32_t valuelen = 0;
@ -1323,7 +1323,7 @@ static SSdbRow *mndUserActionDecode(SSdbRaw *pRaw) {
if (value == NULL) {
TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _OVER);
}
memset(value, 0, valuelen);
(void)memset(value, 0, valuelen);
SDB_GET_BINARY(pRaw, dataPos, value, valuelen, _OVER)
TAOS_CHECK_GOTO(taosHashPut(pUser->readTbs, key, keyLen, value, valuelen), &lino, _OVER);
@ -1337,7 +1337,7 @@ static SSdbRow *mndUserActionDecode(SSdbRaw *pRaw) {
if (key == NULL) {
TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _OVER);
}
memset(key, 0, keyLen);
(void)memset(key, 0, keyLen);
SDB_GET_BINARY(pRaw, dataPos, key, keyLen, _OVER);
int32_t valuelen = 0;
@ -1346,7 +1346,7 @@ static SSdbRow *mndUserActionDecode(SSdbRaw *pRaw) {
if (value == NULL) {
TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _OVER);
}
memset(value, 0, valuelen);
(void)memset(value, 0, valuelen);
SDB_GET_BINARY(pRaw, dataPos, value, valuelen, _OVER)
TAOS_CHECK_GOTO(taosHashPut(pUser->writeTbs, key, keyLen, value, valuelen), &lino, _OVER);
@ -1361,7 +1361,7 @@ static SSdbRow *mndUserActionDecode(SSdbRaw *pRaw) {
if (key == NULL) {
TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _OVER);
}
memset(key, 0, keyLen);
(void)memset(key, 0, keyLen);
SDB_GET_BINARY(pRaw, dataPos, key, keyLen, _OVER);
int32_t valuelen = 0;
@ -1370,7 +1370,7 @@ static SSdbRow *mndUserActionDecode(SSdbRaw *pRaw) {
if (value == NULL) {
TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _OVER);
}
memset(value, 0, valuelen);
(void)memset(value, 0, valuelen);
SDB_GET_BINARY(pRaw, dataPos, value, valuelen, _OVER)
TAOS_CHECK_GOTO(taosHashPut(pUser->alterTbs, key, keyLen, value, valuelen), &lino, _OVER);
@ -1384,7 +1384,7 @@ static SSdbRow *mndUserActionDecode(SSdbRaw *pRaw) {
if (key == NULL) {
TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _OVER);
}
memset(key, 0, keyLen);
(void)memset(key, 0, keyLen);
SDB_GET_BINARY(pRaw, dataPos, key, keyLen, _OVER);
int32_t valuelen = 0;
@ -1393,7 +1393,7 @@ static SSdbRow *mndUserActionDecode(SSdbRaw *pRaw) {
if (value == NULL) {
TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _OVER);
}
memset(value, 0, valuelen);
(void)memset(value, 0, valuelen);
SDB_GET_BINARY(pRaw, dataPos, value, valuelen, _OVER)
TAOS_CHECK_GOTO(taosHashPut(pUser->readViews, key, keyLen, value, valuelen), &lino, _OVER);
@ -1407,7 +1407,7 @@ static SSdbRow *mndUserActionDecode(SSdbRaw *pRaw) {
if (key == NULL) {
TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _OVER);
}
memset(key, 0, keyLen);
(void)memset(key, 0, keyLen);
SDB_GET_BINARY(pRaw, dataPos, key, keyLen, _OVER);
int32_t valuelen = 0;
@ -1416,7 +1416,7 @@ static SSdbRow *mndUserActionDecode(SSdbRaw *pRaw) {
if (value == NULL) {
TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _OVER);
}
memset(value, 0, valuelen);
(void)memset(value, 0, valuelen);
SDB_GET_BINARY(pRaw, dataPos, value, valuelen, _OVER)
TAOS_CHECK_GOTO(taosHashPut(pUser->writeViews, key, keyLen, value, valuelen), &lino, _OVER);
@ -1430,7 +1430,7 @@ static SSdbRow *mndUserActionDecode(SSdbRaw *pRaw) {
if (key == NULL) {
TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _OVER);
}
memset(key, 0, keyLen);
(void)memset(key, 0, keyLen);
SDB_GET_BINARY(pRaw, dataPos, key, keyLen, _OVER);
int32_t valuelen = 0;
@ -1439,7 +1439,7 @@ static SSdbRow *mndUserActionDecode(SSdbRaw *pRaw) {
if (value == NULL) {
TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _OVER);
}
memset(value, 0, valuelen);
(void)memset(value, 0, valuelen);
SDB_GET_BINARY(pRaw, dataPos, value, valuelen, _OVER)
TAOS_CHECK_GOTO(taosHashPut(pUser->alterViews, key, keyLen, value, valuelen), &lino, _OVER);
@ -1454,7 +1454,7 @@ static SSdbRow *mndUserActionDecode(SSdbRaw *pRaw) {
if (key == NULL) {
TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _OVER);
}
memset(key, 0, keyLen);
(void)memset(key, 0, keyLen);
SDB_GET_BINARY(pRaw, dataPos, key, keyLen, _OVER);
int32_t ref = 0;
@ -1645,7 +1645,7 @@ static int32_t mndUserActionUpdate(SSdb *pSdb, SUserObj *pOld, SUserObj *pNew) {
pOld->sysInfo = pNew->sysInfo;
pOld->enable = pNew->enable;
pOld->flag = pNew->flag;
memcpy(pOld->pass, pNew->pass, TSDB_PASSWORD_LEN);
(void)memcpy(pOld->pass, pNew->pass, TSDB_PASSWORD_LEN);
TSWAP(pOld->readDbs, pNew->readDbs);
TSWAP(pOld->writeDbs, pNew->writeDbs);
TSWAP(pOld->topics, pNew->topics);
@ -1663,7 +1663,7 @@ static int32_t mndUserActionUpdate(SSdb *pSdb, SUserObj *pOld, SUserObj *pNew) {
taosWUnLockLatch(&pOld->lock);
return TSDB_CODE_OUT_OF_MEMORY;
}
memcpy(pOld->pIpWhiteList, pNew->pIpWhiteList, sz);
(void)memcpy(pOld->pIpWhiteList, pNew->pIpWhiteList, sz);
pOld->ipWhiteListVer = pNew->ipWhiteListVer;
taosWUnLockLatch(&pOld->lock);
@ -1854,13 +1854,13 @@ static int32_t mndProcessCreateUserReq(SRpcMsg *pReq) {
if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
char detail[1000] = {0};
sprintf(detail, "enable:%d, superUser:%d, sysInfo:%d, password:xxx", createReq.enable, createReq.superUser,
createReq.sysInfo);
(void)sprintf(detail, "enable:%d, superUser:%d, sysInfo:%d, password:xxx", createReq.enable, createReq.superUser,
createReq.sysInfo);
char operation[15] = {0};
if (createReq.isImport == 1) {
strcpy(operation, "importUser");
(void)strcpy(operation, "importUser");
} else {
strcpy(operation, "createUser");
(void)strcpy(operation, "createUser");
}
auditRecord(pReq, pMnode->clusterId, operation, "", createReq.user, detail, strlen(detail));
@ -2037,7 +2037,7 @@ static int32_t mndTablePriviledge(SMnode *pMnode, SHashObj *hash, SHashObj *useD
void *pIter = NULL;
char tbFName[TSDB_TABLE_FNAME_LEN] = {0};
snprintf(tbFName, sizeof(tbFName), "%s.%s", alterReq->objname, alterReq->tabName);
(void)snprintf(tbFName, sizeof(tbFName), "%s.%s", alterReq->objname, alterReq->tabName);
int32_t len = strlen(tbFName) + 1;
if (alterReq->tagCond != NULL && alterReq->tagCondLen != 0) {
@ -2067,7 +2067,7 @@ static int32_t mndRemoveTablePriviledge(SMnode *pMnode, SHashObj *hash, SHashObj
SSdb *pSdb) {
void *pIter = NULL;
char tbFName[TSDB_TABLE_FNAME_LEN] = {0};
snprintf(tbFName, sizeof(tbFName), "%s.%s", alterReq->objname, alterReq->tabName);
(void)snprintf(tbFName, sizeof(tbFName), "%s.%s", alterReq->objname, alterReq->tabName);
int32_t len = strlen(tbFName) + 1;
if (taosHashRemove(hash, tbFName, len) != 0) {
@ -2323,7 +2323,7 @@ static int32_t mndProcessAlterUserReq(SRpcMsg *pReq) {
if (alterReq.alterType == TSDB_ALTER_USER_PASSWD) {
char pass[TSDB_PASSWORD_LEN + 1] = {0};
taosEncryptPass_c((uint8_t *)alterReq.pass, strlen(alterReq.pass), pass);
memcpy(newUser.pass, pass, TSDB_PASSWORD_LEN);
(void)memcpy(newUser.pass, pass, TSDB_PASSWORD_LEN);
if (0 != strncmp(pUser->pass, pass, TSDB_PASSWORD_LEN)) {
++newUser.passVersion;
}
@ -2361,12 +2361,12 @@ static int32_t mndProcessAlterUserReq(SRpcMsg *pReq) {
}
bool exist = false;
memcpy(pNew->pIpRange, pUser->pIpWhiteList->pIpRange, sizeof(SIpV4Range) * idx);
(void)memcpy(pNew->pIpRange, pUser->pIpWhiteList->pIpRange, sizeof(SIpV4Range) * idx);
for (int i = 0; i < alterReq.numIpRanges; i++) {
SIpV4Range *range = &(alterReq.pIpRanges[i]);
if (!isRangeInIpWhiteList(pUser->pIpWhiteList, range)) {
// already exist, just ignore;
memcpy(&pNew->pIpRange[idx], range, sizeof(SIpV4Range));
(void)memcpy(&pNew->pIpRange[idx], range, sizeof(SIpV4Range));
idx++;
continue;
} else {
@ -2416,7 +2416,7 @@ static int32_t mndProcessAlterUserReq(SRpcMsg *pReq) {
if (localHost) break;
if (found == false) {
memcpy(&pNew->pIpRange[idx], oldRange, sizeof(SIpV4Range));
(void)memcpy(&pNew->pIpRange[idx], oldRange, sizeof(SIpV4Range));
idx++;
} else {
noexist = false;
@ -2445,9 +2445,9 @@ static int32_t mndProcessAlterUserReq(SRpcMsg *pReq) {
if (alterReq.alterType == TSDB_ALTER_USER_PASSWD) {
char detail[1000] = {0};
sprintf(detail, "alterType:%s, enable:%d, superUser:%d, sysInfo:%d, createdb:%d, tabName:%s, password:xxx",
mndUserAuditTypeStr(alterReq.alterType), alterReq.enable, alterReq.superUser, alterReq.sysInfo,
alterReq.createdb ? 1 : 0, alterReq.tabName);
(void)sprintf(detail, "alterType:%s, enable:%d, superUser:%d, sysInfo:%d, createdb:%d, tabName:%s, password:xxx",
mndUserAuditTypeStr(alterReq.alterType), alterReq.enable, alterReq.superUser, alterReq.sysInfo,
alterReq.createdb ? 1 : 0, alterReq.tabName);
auditRecord(pReq, pMnode->clusterId, "alterUser", "", alterReq.user, detail, strlen(detail));
} else if (alterReq.alterType == TSDB_ALTER_USER_SUPERUSER || alterReq.alterType == TSDB_ALTER_USER_ENABLE ||
alterReq.alterType == TSDB_ALTER_USER_SYSINFO || alterReq.alterType == TSDB_ALTER_USER_CREATEDB) {
@ -2460,7 +2460,7 @@ static int32_t mndProcessAlterUserReq(SRpcMsg *pReq) {
ALTER_USER_ADD_ALL_TB_PRIV(alterReq.alterType, alterReq.privileges, alterReq.tabName)) {
if (strcmp(alterReq.objname, "1.*") != 0) {
SName name = {0};
tNameFromString(&name, alterReq.objname, T_NAME_ACCT | T_NAME_DB);
(void)tNameFromString(&name, alterReq.objname, T_NAME_ACCT | T_NAME_DB);
auditRecord(pReq, pMnode->clusterId, "GrantPrivileges", name.dbname, alterReq.user, alterReq.sql,
alterReq.sqlLen);
} else {
@ -2475,7 +2475,7 @@ static int32_t mndProcessAlterUserReq(SRpcMsg *pReq) {
} else {
if (strcmp(alterReq.objname, "1.*") != 0) {
SName name = {0};
tNameFromString(&name, alterReq.objname, T_NAME_ACCT | T_NAME_DB);
(void)tNameFromString(&name, alterReq.objname, T_NAME_ACCT | T_NAME_DB);
auditRecord(pReq, pMnode->clusterId, "RevokePrivileges", name.dbname, alterReq.user, alterReq.sql,
alterReq.sqlLen);
} else {
@ -2658,7 +2658,7 @@ static int32_t mndRetrieveUsers(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBl
TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _exit);
}
varDataSetLen(varstr, tlen);
memcpy(varDataVal(varstr), buf, tlen);
(void)memcpy(varDataVal(varstr), buf, tlen);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols);
COL_DATA_SET_VAL_GOTO((const char *)varstr, false, pUser, _exit);
@ -2743,7 +2743,7 @@ static int32_t mndRetrieveUsersFull(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock
TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _exit);
}
varDataSetLen(varstr, tlen);
memcpy(varDataVal(varstr), buf, tlen);
(void)memcpy(varDataVal(varstr), buf, tlen);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols);
COL_DATA_SET_VAL_GOTO((const char *)varstr, false, pUser, _exit);
@ -2834,7 +2834,7 @@ static int32_t mndLoopHash(SHashObj *hash, char *priType, SSDataBlock *pBlock, i
nodesDestroyNode(pAst);
} else {
sqlLen = 5;
sprintf(*sql, "error");
(void)sprintf(*sql, "error");
}
STR_WITH_MAXSIZE_TO_VARSTR((*condition), (*sql), pShow->pMeta->pSchemas[cols].bytes);
@ -2978,8 +2978,8 @@ static int32_t mndRetrievePrivileges(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock
SName name = {0};
char objName[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
tNameFromString(&name, db, T_NAME_ACCT | T_NAME_DB);
tNameGetDbName(&name, varDataVal(objName));
(void)tNameFromString(&name, db, T_NAME_ACCT | T_NAME_DB);
(void)tNameGetDbName(&name, varDataVal(objName));
varDataSetLen(objName, strlen(varDataVal(objName)));
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
COL_DATA_SET_VAL_GOTO((const char *)objName, false, pUser, _exit);
@ -3022,8 +3022,8 @@ static int32_t mndRetrievePrivileges(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock
SName name = {0};
char objName[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
tNameFromString(&name, db, T_NAME_ACCT | T_NAME_DB);
tNameGetDbName(&name, varDataVal(objName));
(void)tNameFromString(&name, db, T_NAME_ACCT | T_NAME_DB);
(void)tNameGetDbName(&name, varDataVal(objName));
varDataSetLen(objName, strlen(varDataVal(objName)));
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
COL_DATA_SET_VAL_GOTO((const char *)objName, false, pUser, _exit);
@ -3143,7 +3143,7 @@ int32_t mndValidateUserAuthInfo(SMnode *pMnode, SUserAuthVersion *pUsers, int32_
if (pUser == NULL) {
if (TSDB_CODE_MND_USER_NOT_EXIST == code) {
SGetUserAuthRsp rsp = {.dropped = 1};
memcpy(rsp.user, pUsers[i].user, TSDB_USER_LEN);
(void)memcpy(rsp.user, pUsers[i].user, TSDB_USER_LEN);
(void)taosArrayPush(batchRsp.pArray, &rsp);
}
mError("user:%s, failed to auth user since %s", pUsers[i].user, terrstr());

22
source/dnode/vnode/src/sma/smaEnv.c
View File

@ -103,7 +103,7 @@ void smaCleanUp() {
}
if (old == 1) {
taosCloseRef(smaMgmt.rsetId);
(void)taosCloseRef(smaMgmt.rsetId);
taosHashCleanup(smaMgmt.refHash);
smaMgmt.refHash = NULL;
taosTmrCleanUp(smaMgmt.tmrHandle);
@ -174,10 +174,10 @@ static void tRSmaInfoHashFreeNode(void *data) {
if ((pRSmaInfo = *(SRSmaInfo **)data)) {
if ((pItem = RSMA_INFO_ITEM((SRSmaInfo *)pRSmaInfo, 0)) && pItem->level) {
taosHashRemove(smaMgmt.refHash, &pItem, POINTER_BYTES);
(void)taosHashRemove(smaMgmt.refHash, &pItem, POINTER_BYTES);
}
if ((pItem = RSMA_INFO_ITEM((SRSmaInfo *)pRSmaInfo, 1)) && pItem->level) {
taosHashRemove(smaMgmt.refHash, &pItem, POINTER_BYTES);
(void)taosHashRemove(smaMgmt.refHash, &pItem, POINTER_BYTES);
}
tdFreeRSmaInfo(pRSmaInfo->pSma, pRSmaInfo);
}
@ -378,16 +378,16 @@ int32_t tdCheckAndInitSmaEnv(SSma *pSma, int8_t smaType) {
}
// init sma env
tdLockSma(pSma);
(void)tdLockSma(pSma);
pEnv = (smaType == TSDB_SMA_TYPE_TIME_RANGE) ? atomic_load_ptr(&SMA_TSMA_ENV(pSma))
: atomic_load_ptr(&SMA_RSMA_ENV(pSma));
if (!pEnv) {
if ((code = tdInitSmaEnv(pSma, smaType, &pEnv)) < 0) {
tdUnLockSma(pSma);
(void)tdUnLockSma(pSma);
TAOS_RETURN(code);
}
}
tdUnLockSma(pSma);
(void)tdUnLockSma(pSma);
TAOS_RETURN(TSDB_CODE_SUCCESS);
}
@ -395,15 +395,15 @@ int32_t tdCheckAndInitSmaEnv(SSma *pSma, int8_t smaType) {
void *tdRSmaExecutorFunc(void *param) {
setThreadName("vnode-rsma");
tdRSmaProcessExecImpl((SSma *)param, RSMA_EXEC_OVERFLOW);
(void)tdRSmaProcessExecImpl((SSma *)param, RSMA_EXEC_OVERFLOW);
return NULL;
}
static int32_t tdRsmaStartExecutor(const SSma *pSma) {
int32_t code = 0;
TdThreadAttr thAttr = {0};
taosThreadAttrInit(&thAttr);
taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);
(void)taosThreadAttrInit(&thAttr);
(void)taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);
SSmaEnv *pEnv = SMA_RSMA_ENV(pSma);
SSmaStat *pStat = SMA_ENV_STAT(pEnv);
@ -418,7 +418,7 @@ static int32_t tdRsmaStartExecutor(const SSma *pSma) {
smaDebug("vgId:%d, success to create pthread for rsma", SMA_VID(pSma));
}
taosThreadAttrDestroy(&thAttr);
(void)taosThreadAttrDestroy(&thAttr);
TAOS_RETURN(code);
}
@ -444,7 +444,7 @@ static int32_t tdRsmaStopExecutor(const SSma *pSma) {
for (int32_t i = 0; i < tsNumOfVnodeRsmaThreads; ++i) {
if (taosCheckPthreadValid(pthread[i])) {
smaDebug("vgId:%d, start to join pthread for rsma:%" PRId64 "", SMA_VID(pSma), taosGetPthreadId(pthread[i]));
taosThreadJoin(pthread[i], NULL);
(void)taosThreadJoin(pthread[i], NULL);
}
}

6
source/dnode/vnode/src/sma/smaOpen.c
View File

@ -145,7 +145,7 @@ int32_t smaOpen(SVnode *pVnode, int8_t rollback, bool force) {
pVnode->pSma = pSma;
pSma->pVnode = pVnode;
taosThreadMutexInit(&pSma->mutex, NULL);
(void)taosThreadMutexInit(&pSma->mutex, NULL);
pSma->locked = false;
if (VND_IS_RSMA(pVnode)) {
@ -173,8 +173,8 @@ _exit:
int32_t smaClose(SSma *pSma) {
if (pSma) {
smaPreClose(pSma);
taosThreadMutexDestroy(&pSma->mutex);
(void)smaPreClose(pSma);
(void)taosThreadMutexDestroy(&pSma->mutex);
SMA_TSMA_ENV(pSma) = tdFreeSmaEnv(SMA_TSMA_ENV(pSma));
SMA_RSMA_ENV(pSma) = tdFreeSmaEnv(SMA_RSMA_ENV(pSma));
if SMA_RSMA_TSDB0 (pSma) tsdbClose(&SMA_RSMA_TSDB0(pSma));

34
source/dnode/vnode/src/sma/smaRollup.c
View File

@ -89,7 +89,7 @@ void *tdFreeRSmaInfo(SSma *pSma, SRSmaInfo *pInfo) {
if (pItem->tmrId) {
smaDebug("vgId:%d, stop fetch timer %p for table %" PRIi64 " level %d", SMA_VID(pSma), pItem->tmrId,
pInfo->suid, i + 1);
taosTmrStopA(&pItem->tmrId);
(void)taosTmrStopA(&pItem->tmrId);
}
if (pItem->pStreamState) {
@ -246,7 +246,7 @@ static void tdRSmaTaskInit(SStreamMeta *pMeta, SRSmaInfoItem *pItem, SStreamTask
}
static void tdRSmaTaskRemove(SStreamMeta *pMeta, int64_t streamId, int32_t taskId) {
streamMetaUnregisterTask(pMeta, streamId, taskId);
(void)streamMetaUnregisterTask(pMeta, streamId, taskId);
streamMetaWLock(pMeta);
int32_t numOfTasks = streamMetaGetNumOfTasks(pMeta);
if (streamMetaCommit(pMeta) < 0) {
@ -296,7 +296,7 @@ static int32_t tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaStat
if (!pStreamTask->exec.qmsg) {
TAOS_RETURN(TSDB_CODE_OUT_OF_MEMORY);
}
sprintf(pStreamTask->exec.qmsg, "%s", RSMA_EXEC_TASK_FLAG);
(void)sprintf(pStreamTask->exec.qmsg, "%s", RSMA_EXEC_TASK_FLAG);
pStreamTask->chkInfo.checkpointId = streamMetaGetLatestCheckpointId(pStreamTask->pMeta);
tdRSmaTaskInit(pStreamTask->pMeta, pItem, &pStreamTask->id);
@ -347,7 +347,7 @@ static int32_t tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaStat
TAOS_RETURN(TSDB_CODE_OUT_OF_MEMORY);
}
taosTmrReset(tdRSmaFetchTrigger, RSMA_FETCH_INTERVAL, pItem, smaMgmt.tmrHandle, &pItem->tmrId);
(void)taosTmrReset(tdRSmaFetchTrigger, RSMA_FETCH_INTERVAL, pItem, smaMgmt.tmrHandle, &pItem->tmrId);
smaInfo("vgId:%d, open rsma task:%p table:%" PRIi64 " level:%" PRIi8 ", checkpointId:%" PRIi64
", submitReqVer:%" PRIi64 ", fetchResultVer:%" PRIi64 ", maxdelay:%" PRIi64 " watermark:%" PRIi64
@ -799,7 +799,7 @@ static int32_t tdExecuteRSmaImplAsync(SSma *pSma, int64_t version, const void *p
*(int32_t *)pItem = len;
pItem = POINTER_SHIFT(pItem, sizeof(int32_t));
*(int64_t *)pItem = version;
memcpy(POINTER_SHIFT(pItem, sizeof(int64_t)), pMsg, len);
(void)memcpy(POINTER_SHIFT(pItem, sizeof(int64_t)), pMsg, len);
TAOS_CHECK_RETURN(taosWriteQitem(pInfo->queue, qItem));
@ -810,7 +810,7 @@ static int32_t tdExecuteRSmaImplAsync(SSma *pSma, int64_t version, const void *p
int64_t nItems = atomic_fetch_add_64(&pRSmaStat->nBufItems, 1);
if (atomic_load_8(&pInfo->assigned) == 0) {
tsem_post(&(pRSmaStat->notEmpty));
(void)tsem_post(&(pRSmaStat->notEmpty));
}
// smoothing consume
@ -1339,7 +1339,7 @@ static void tdRSmaFetchTrigger(void *param, void *tmrId) {
if (!(pStat = (SRSmaStat *)tdAcquireSmaRef(smaMgmt.rsetId, pRSmaRef->refId))) {
smaWarn("rsma fetch task not start since rsma stat already destroyed, rsetId:%d refId:%" PRIi64 ")", smaMgmt.rsetId,
pRSmaRef->refId); // pRSmaRef freed in taosHashRemove
taosHashRemove(smaMgmt.refHash, &param, POINTER_BYTES);
(void)taosHashRemove(smaMgmt.refHash, &param, POINTER_BYTES);
return;
}
@ -1348,8 +1348,8 @@ static void tdRSmaFetchTrigger(void *param, void *tmrId) {
if ((code = tdAcquireRSmaInfoBySuid(pSma, pRSmaRef->suid, &pRSmaInfo)) != 0) {
smaDebug("rsma fetch task not start since rsma info not exist, rsetId:%d refId:%" PRIi64 ")", smaMgmt.rsetId,
pRSmaRef->refId); // pRSmaRef freed in taosHashRemove
tdReleaseSmaRef(smaMgmt.rsetId, pRSmaRef->refId);
taosHashRemove(smaMgmt.refHash, &param, POINTER_BYTES);
(void)tdReleaseSmaRef(smaMgmt.rsetId, pRSmaRef->refId);
(void)taosHashRemove(smaMgmt.refHash, &param, POINTER_BYTES);
return;
}
@ -1357,8 +1357,8 @@ static void tdRSmaFetchTrigger(void *param, void *tmrId) {
smaDebug("rsma fetch task not start since rsma info already deleted, rsetId:%d refId:%" PRIi64 ")", smaMgmt.rsetId,
pRSmaRef->refId); // pRSmaRef freed in taosHashRemove
tdReleaseRSmaInfo(pSma, pRSmaInfo);
tdReleaseSmaRef(smaMgmt.rsetId, pRSmaRef->refId);
taosHashRemove(smaMgmt.refHash, &param, POINTER_BYTES);
(void)tdReleaseSmaRef(smaMgmt.rsetId, pRSmaRef->refId);
(void)taosHashRemove(smaMgmt.refHash, &param, POINTER_BYTES);
return;
}
@ -1376,7 +1376,7 @@ static void tdRSmaFetchTrigger(void *param, void *tmrId) {
taosTmrReset(tdRSmaFetchTrigger, RSMA_FETCH_INTERVAL, pItem, smaMgmt.tmrHandle, &pItem->tmrId);
}
tdReleaseRSmaInfo(pSma, pRSmaInfo);
tdReleaseSmaRef(smaMgmt.rsetId, pRSmaRef->refId);
(void)tdReleaseSmaRef(smaMgmt.rsetId, pRSmaRef->refId);
return;
}
default:
@ -1393,7 +1393,7 @@ static void tdRSmaFetchTrigger(void *param, void *tmrId) {
atomic_store_8(&pItem->fetchLevel, 1);
if (atomic_load_8(&pRSmaInfo->assigned) == 0) {
tsem_post(&(pStat->notEmpty));
(void)tsem_post(&(pStat->notEmpty));
}
} break;
case TASK_TRIGGER_STAT_INACTIVE: {
@ -1414,7 +1414,7 @@ static void tdRSmaFetchTrigger(void *param, void *tmrId) {
_end:
taosTmrReset(tdRSmaFetchTrigger, pItem->maxDelay, pItem, smaMgmt.tmrHandle, &pItem->tmrId);
tdReleaseRSmaInfo(pSma, pRSmaInfo);
tdReleaseSmaRef(smaMgmt.rsetId, pRSmaRef->refId);
(void)tdReleaseSmaRef(smaMgmt.rsetId, pRSmaRef->refId);
}
static void tdFreeRSmaSubmitItems(SArray *pItems, int32_t type) {
@ -1507,7 +1507,7 @@ static int32_t tdRSmaBatchExec(SSma *pSma, SRSmaInfo *pInfo, STaosQall *qall, SA
// the submitReq/deleteReq msg may exsit alternately in the msg queue, consume them sequentially in batch mode
while (1) {
taosGetQitem(qall, (void **)&msg);
(void)taosGetQitem(qall, (void **)&msg);
if (msg) {
int8_t inputType = RSMA_EXEC_MSG_TYPE(msg);
if (inputType == STREAM_INPUT__DATA_SUBMIT) {
@ -1574,7 +1574,7 @@ _exit:
tdFreeRSmaSubmitItems(pSubmitArr, nSubmit ? STREAM_INPUT__MERGED_SUBMIT : STREAM_INPUT__REF_DATA_BLOCK);
while (1) {
void *msg = NULL;
taosGetQitem(qall, (void **)&msg);
(void)taosGetQitem(qall, (void **)&msg);
if (msg) {
taosFreeQitem(msg);
} else {
@ -1687,7 +1687,7 @@ int32_t tdRSmaProcessExecImpl(SSma *pSma, ERsmaExecType type) {
break;
}
tsem_wait(&pRSmaStat->notEmpty);
(void)tsem_wait(&pRSmaStat->notEmpty);
if ((pEnv->flag & SMA_ENV_FLG_CLOSE) && (atomic_load_64(&pRSmaStat->nBufItems) <= 0)) {
smaDebug("vgId:%d, exec task end, flag:%" PRIi8 ", nBufItems:%" PRIi64, SMA_VID(pSma), pEnv->flag,

4
source/dnode/vnode/src/sma/smaSnapshot.c
View File

@ -93,7 +93,7 @@ int32_t rsmaSnapRead(SRSmaSnapReader* pReader, uint8_t** ppData) {
_exit:
if (code) {
smaError("vgId:%d, %s failed at line %d since %s", SMA_VID(pReader->pSma), __func__, lino, tstrerror(code));
rsmaSnapReaderClose(&pReader);
(void)rsmaSnapReaderClose(&pReader);
} else {
smaInfo("vgId:%d, vnode snapshot rsma read succeed", SMA_VID(pReader->pSma));
}
@ -106,7 +106,7 @@ int32_t rsmaSnapReaderClose(SRSmaSnapReader** ppReader) {
for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
if (pReader->pDataReader[i]) {
tsdbSnapReaderClose(&pReader->pDataReader[i]);
(void)tsdbSnapReaderClose(&pReader->pDataReader[i]);
}
}

5
source/dnode/vnode/src/sma/smaTimeRange.c
View File

@ -204,7 +204,10 @@ int32_t smaBlockToSubmit(SVnode *pVnode, const SArray *pBlocks, const STSchema *
continue;
}
taosArrayPush(pReq->aSubmitTbData, &tbData);
if( taosArrayPush(pReq->aSubmitTbData, &tbData) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
continue;
}
int32_t size = (int32_t)taosArrayGetSize(pReq->aSubmitTbData) - 1;
TAOS_CHECK_EXIT(taosHashPut(pTableIndexMap, &groupId, sizeof(groupId), &size, sizeof(size)));

8
source/dnode/vnode/src/sma/smaUtil.c
View File

@ -19,19 +19,19 @@
void tdRSmaQTaskInfoGetFullPath(SVnode *pVnode, tb_uid_t suid, int8_t level, STfs *pTfs, char *outputName) {
tdRSmaGetDirName(pVnode, pTfs, true, outputName);
int32_t rsmaLen = strlen(outputName);
snprintf(outputName + rsmaLen, TSDB_FILENAME_LEN - rsmaLen, "%" PRIi8 "%s%" PRIi64, level, TD_DIRSEP, suid);
(void)snprintf(outputName + rsmaLen, TSDB_FILENAME_LEN - rsmaLen, "%" PRIi8 "%s%" PRIi64, level, TD_DIRSEP, suid);
}
void tdRSmaGetDirName(SVnode *pVnode, STfs *pTfs, bool endWithSep, char *outputName) {
int32_t offset = 0;
// vnode
vnodeGetPrimaryDir(pVnode->path, pVnode->diskPrimary, pTfs, outputName, TSDB_FILENAME_LEN);
(void)vnodeGetPrimaryDir(pVnode->path, pVnode->diskPrimary, pTfs, outputName, TSDB_FILENAME_LEN);
offset = strlen(outputName);
// rsma
snprintf(outputName + offset, TSDB_FILENAME_LEN - offset - 1, "%s%s%s", TD_DIRSEP, VNODE_RSMA_DIR,
(endWithSep ? TD_DIRSEP : ""));
(void)snprintf(outputName + offset, TSDB_FILENAME_LEN - offset - 1, "%s%s%s", TD_DIRSEP, VNODE_RSMA_DIR,
(endWithSep ? TD_DIRSEP : ""));
}
// smaXXXUtil ================

17
source/libs/sync/inc/syncPipeline.h
View File

@ -77,7 +77,7 @@ static FORCE_INLINE int32_t syncLogReplGetNextRetryBackoff(SSyncLogReplMgr* pMgr
return TMIN(pMgr->retryBackoff + 1, SYNC_MAX_RETRY_BACKOFF);
}
SyncTerm syncLogReplGetPrevLogTerm(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncIndex index);
int32_t syncLogReplGetPrevLogTerm(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncIndex index, SyncTerm* pSyncTerm);
int32_t syncLogReplStart(SSyncLogReplMgr* pMgr, SSyncNode* pNode);
int32_t syncLogReplAttempt(SSyncLogReplMgr* pMgr, SSyncNode* pNode);
@ -93,10 +93,10 @@ int32_t syncLogReplContinue(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncAppendE
int32_t syncLogReplProcessHeartbeatReply(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncHeartbeatReply* pMsg);
// SSyncLogBuffer
SSyncLogBuffer* syncLogBufferCreate();
void syncLogBufferDestroy(SSyncLogBuffer* pBuf);
int32_t syncLogBufferInit(SSyncLogBuffer* pBuf, SSyncNode* pNode);
int32_t syncLogBufferReInit(SSyncLogBuffer* pBuf, SSyncNode* pNode);
int32_t syncLogBufferCreate(SSyncLogBuffer** ppBuf);
void syncLogBufferDestroy(SSyncLogBuffer* pBuf);
int32_t syncLogBufferInit(SSyncLogBuffer* pBuf, SSyncNode* pNode);
int32_t syncLogBufferReInit(SSyncLogBuffer* pBuf, SSyncNode* pNode);
// access
int64_t syncLogBufferGetEndIndex(SSyncLogBuffer* pBuf);
@ -110,9 +110,10 @@ int32_t syncLogBufferCommit(SSyncLogBuffer* pBuf, SSyncNode* pNode, int64_t comm
int32_t syncLogBufferReset(SSyncLogBuffer* pBuf, SSyncNode* pNode);
// private
SSyncRaftEntry* syncLogBufferGetOneEntry(SSyncLogBuffer* pBuf, SSyncNode* pNode, SyncIndex index, bool* pInBuf);
int32_t syncLogBufferValidate(SSyncLogBuffer* pBuf);
int32_t syncLogBufferRollback(SSyncLogBuffer* pBuf, SSyncNode* pNode, SyncIndex toIndex);
int32_t syncLogBufferGetOneEntry(SSyncLogBuffer* pBuf, SSyncNode* pNode, SyncIndex index, bool* pInBuf,
SSyncRaftEntry** ppEntry);
int32_t syncLogBufferValidate(SSyncLogBuffer* pBuf);
int32_t syncLogBufferRollback(SSyncLogBuffer* pBuf, SSyncNode* pNode, SyncIndex toIndex);
int32_t syncFsmExecute(SSyncNode* pNode, SSyncFSM* pFsm, ESyncState role, SyncTerm term, SSyncRaftEntry* pEntry,
int32_t applyCode, bool force);

2
source/libs/sync/inc/syncRaftEntry.h
View File

@ -43,7 +43,7 @@ SSyncRaftEntry* syncEntryBuildFromRpcMsg(const SRpcMsg* pMsg, SyncTerm term, Syn
SSyncRaftEntry* syncEntryBuildFromAppendEntries(const SyncAppendEntries* pMsg);
SSyncRaftEntry* syncEntryBuildNoop(SyncTerm term, SyncIndex index, int32_t vgId);
void syncEntryDestroy(SSyncRaftEntry* pEntry);
void syncEntry2OriginalRpc(const SSyncRaftEntry* pEntry, SRpcMsg* pRpcMsg); // step 7
int32_t syncEntry2OriginalRpc(const SSyncRaftEntry* pEntry, SRpcMsg* pRpcMsg); // step 7
static FORCE_INLINE bool syncLogReplBarrier(SSyncRaftEntry* pEntry) {
return pEntry->originalRpcType == TDMT_SYNC_NOOP || pEntry->originalRpcType == TDMT_SYNC_CONFIG_CHANGE;

2
source/libs/sync/inc/syncReplication.h
View File

@ -58,7 +58,7 @@ int32_t syncNodeSendAppendEntries(SSyncNode* pNode, const SRaftId* destRaftId, S
int32_t syncSnapSendMsg(SSyncSnapshotSender* pSender, int32_t seq, void* pBlock, int32_t len, int32_t typ);
int32_t syncSnapSendRsp(SSyncSnapshotReceiver* pReceiver, SyncSnapshotSend* pMsg, void* pBlock, int32_t len,
int32_t typ, int32_t code);
int32_t typ, int32_t rspCode);
#ifdef __cplusplus
}

16
source/libs/sync/inc/syncRespMgr.h
View File

@ -35,14 +35,14 @@ typedef struct SSyncRespMgr {
uint64_t seqNum;
} SSyncRespMgr;
SSyncRespMgr *syncRespMgrCreate(void *data, int64_t ttl);
void syncRespMgrDestroy(SSyncRespMgr *pObj);
uint64_t syncRespMgrAdd(SSyncRespMgr *pObj, const SRespStub *pStub);
int32_t syncRespMgrDel(SSyncRespMgr *pObj, uint64_t seq);
int32_t syncRespMgrGet(SSyncRespMgr *pObj, uint64_t seq, SRespStub *pStub);
int32_t syncRespMgrGetAndDel(SSyncRespMgr *pObj, uint64_t seq, SRpcHandleInfo *pInfo);
void syncRespClean(SSyncRespMgr *pObj);
void syncRespCleanRsp(SSyncRespMgr *pObj);
int32_t syncRespMgrCreate(void *data, int64_t ttl, SSyncRespMgr **ppObj);
void syncRespMgrDestroy(SSyncRespMgr *pObj);
uint64_t syncRespMgrAdd(SSyncRespMgr *pObj, const SRespStub *pStub);
int32_t syncRespMgrDel(SSyncRespMgr *pObj, uint64_t seq);
int32_t syncRespMgrGet(SSyncRespMgr *pObj, uint64_t seq, SRespStub *pStub);
int32_t syncRespMgrGetAndDel(SSyncRespMgr *pObj, uint64_t seq, SRpcHandleInfo *pInfo);
void syncRespClean(SSyncRespMgr *pObj);
void syncRespCleanRsp(SSyncRespMgr *pObj);
#ifdef __cplusplus
}

22
source/libs/sync/inc/syncSnapshot.h
View File

@ -74,12 +74,12 @@ typedef struct SSyncSnapshotSender {
int32_t replicaIndex;
} SSyncSnapshotSender;
SSyncSnapshotSender *snapshotSenderCreate(SSyncNode *pSyncNode, int32_t replicaIndex);
void snapshotSenderDestroy(SSyncSnapshotSender *pSender);
bool snapshotSenderIsStart(SSyncSnapshotSender *pSender);
int32_t snapshotSenderStart(SSyncSnapshotSender *pSender);
void snapshotSenderStop(SSyncSnapshotSender *pSender, bool finish);
int32_t snapshotReSend(SSyncSnapshotSender *pSender);
int32_t snapshotSenderCreate(SSyncNode *pSyncNode, int32_t replicaIndex, SSyncSnapshotSender **ppSender);
void snapshotSenderDestroy(SSyncSnapshotSender *pSender);
bool snapshotSenderIsStart(SSyncSnapshotSender *pSender);
int32_t snapshotSenderStart(SSyncSnapshotSender *pSender);
void snapshotSenderStop(SSyncSnapshotSender *pSender, bool finish);
int32_t snapshotReSend(SSyncSnapshotSender *pSender);
typedef struct SSyncSnapshotReceiver {
// update when prep snapshot
@ -101,11 +101,11 @@ typedef struct SSyncSnapshotReceiver {
SSyncNode *pSyncNode;
} SSyncSnapshotReceiver;
SSyncSnapshotReceiver *snapshotReceiverCreate(SSyncNode *pSyncNode, SRaftId fromId);
void snapshotReceiverDestroy(SSyncSnapshotReceiver *pReceiver);
void snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pBeginMsg);
void snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver);
bool snapshotReceiverIsStart(SSyncSnapshotReceiver *pReceiver);
int32_t snapshotReceiverCreate(SSyncNode *pSyncNode, SRaftId fromId, SSyncSnapshotReceiver **ppReceiver);
void snapshotReceiverDestroy(SSyncSnapshotReceiver *pReceiver);
void snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pBeginMsg);
void snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver);
bool snapshotReceiverIsStart(SSyncSnapshotReceiver *pReceiver);
// on message
// int32_t syncNodeOnSnapshot(SSyncNode *ths, const SRpcMsg *pMsg);

17
source/libs/sync/src/syncMain.c
View File

@ -949,6 +949,7 @@ int32_t syncNodeLogStoreRestoreOnNeed(SSyncNode* pNode) {
// open/close --------------
SSyncNode* syncNodeOpen(SSyncInfo* pSyncInfo, int32_t vnodeVersion) {
int32_t code = 0;
SSyncNode* pSyncNode = taosMemoryCalloc(1, sizeof(SSyncNode));
if (pSyncNode == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
@ -1041,7 +1042,7 @@ SSyncNode* syncNodeOpen(SSyncInfo* pSyncInfo, int32_t vnodeVersion) {
pSyncNode->syncEqCtrlMsg = pSyncInfo->syncEqCtrlMsg;
// create raft log ring buffer
pSyncNode->pLogBuf = syncLogBufferCreate();
(void)syncLogBufferCreate(&pSyncNode->pLogBuf); // TODO: check return value
if (pSyncNode->pLogBuf == NULL) {
sError("failed to init sync log buffer since %s. vgId:%d", terrstr(), pSyncNode->vgId);
goto _error;
@ -1218,7 +1219,7 @@ SSyncNode* syncNodeOpen(SSyncInfo* pSyncInfo, int32_t vnodeVersion) {
}
// tools
pSyncNode->pSyncRespMgr = syncRespMgrCreate(pSyncNode, SYNC_RESP_TTL_MS);
(void)syncRespMgrCreate(pSyncNode, SYNC_RESP_TTL_MS, &pSyncNode->pSyncRespMgr); // TODO: check return value
if (pSyncNode->pSyncRespMgr == NULL) {
sError("vgId:%d, failed to create SyncRespMgr", pSyncNode->vgId);
goto _error;
@ -1229,7 +1230,8 @@ SSyncNode* syncNodeOpen(SSyncInfo* pSyncInfo, int32_t vnodeVersion) {
// snapshot senders
for (int32_t i = 0; i < TSDB_MAX_REPLICA + TSDB_MAX_LEARNER_REPLICA; ++i) {
SSyncSnapshotSender* pSender = snapshotSenderCreate(pSyncNode, i);
SSyncSnapshotSender* pSender = NULL;
code = snapshotSenderCreate(pSyncNode, i, &pSender);
if (pSender == NULL) return NULL;
pSyncNode->senders[i] = pSender;
@ -1237,7 +1239,7 @@ SSyncNode* syncNodeOpen(SSyncInfo* pSyncInfo, int32_t vnodeVersion) {
}
// snapshot receivers
pSyncNode->pNewNodeReceiver = snapshotReceiverCreate(pSyncNode, EMPTY_RAFT_ID);
code = snapshotReceiverCreate(pSyncNode, EMPTY_RAFT_ID, &pSyncNode->pNewNodeReceiver);
if (pSyncNode->pNewNodeReceiver == NULL) return NULL;
sRDebug(pSyncNode->pNewNodeReceiver, "snapshot receiver create while open sync node, data:%p",
pSyncNode->pNewNodeReceiver);
@ -1810,7 +1812,7 @@ void syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* pNewConfig, SyncInde
// create new
for (int32_t i = 0; i < TSDB_MAX_REPLICA + TSDB_MAX_LEARNER_REPLICA; ++i) {
if (pSyncNode->senders[i] == NULL) {
pSyncNode->senders[i] = snapshotSenderCreate(pSyncNode, i);
snapshotSenderCreate(pSyncNode, i, &pSyncNode->senders[i]);
if (pSyncNode->senders[i] == NULL) {
// will be created later while send snapshot
sSError(pSyncNode->senders[i], "snapshot sender create failed while reconfig");
@ -2840,8 +2842,9 @@ int32_t syncNodeRebuildAndCopyIfExist(SSyncNode* ths, int32_t oldtotalReplicaNum
}
for (int32_t i = 0; i < TSDB_MAX_REPLICA + TSDB_MAX_LEARNER_REPLICA; ++i) {
SSyncSnapshotSender* pSender = snapshotSenderCreate(ths, i);
if (pSender == NULL) return -1;
SSyncSnapshotSender* pSender = NULL;
int32_t code = snapshotSenderCreate(ths, i, &pSender);
if (pSender == NULL) return terrno = code;
ths->senders[i] = pSender;
sSDebug(pSender, "snapshot sender create while open sync node, data:%p", pSender);

505
source/libs/sync/src/syncPipeline.c
View File

File diff suppressed because it is too large Load Diff

213
source/libs/sync/src/syncRaftCfg.c
View File

@ -18,7 +18,7 @@
#include "syncUtil.h"
#include "tjson.h"
const char* syncRoleToStr(ESyncRole role) {
const char *syncRoleToStr(ESyncRole role) {
switch (role) {
case TAOS_SYNC_ROLE_VOTER:
return "true";
@ -29,11 +29,11 @@ const char* syncRoleToStr(ESyncRole role) {
}
}
const ESyncRole syncStrToRole(char* str) {
if(strcmp(str, "true") == 0){
const ESyncRole syncStrToRole(char *str) {
if (strcmp(str, "true") == 0) {
return TAOS_SYNC_ROLE_VOTER;
}
if(strcmp(str, "false") == 0){
if (strcmp(str, "false") == 0) {
return TAOS_SYNC_ROLE_LEARNER;
}
@ -42,92 +42,130 @@ const ESyncRole syncStrToRole(char* str) {
static int32_t syncEncodeSyncCfg(const void *pObj, SJson *pJson) {
SSyncCfg *pCfg = (SSyncCfg *)pObj;
if (tjsonAddDoubleToObject(pJson, "replicaNum", pCfg->replicaNum) < 0) return -1;
if (tjsonAddDoubleToObject(pJson, "myIndex", pCfg->myIndex) < 0) return -1;
if (tjsonAddDoubleToObject(pJson, "changeVersion", pCfg->changeVersion) < 0) return -1;
int32_t code = 0, lino = 0;
TAOS_CHECK_EXIT(tjsonAddDoubleToObject(pJson, "replicaNum", pCfg->replicaNum));
TAOS_CHECK_EXIT(tjsonAddDoubleToObject(pJson, "myIndex", pCfg->myIndex));
TAOS_CHECK_EXIT(tjsonAddDoubleToObject(pJson, "changeVersion", pCfg->changeVersion));
SJson *nodeInfo = tjsonCreateArray();
if (nodeInfo == NULL) return -1;
if (tjsonAddItemToObject(pJson, "nodeInfo", nodeInfo) < 0) return -1;
if (nodeInfo == NULL) {
TAOS_CHECK_EXIT(TSDB_CODE_OUT_OF_MEMORY);
}
if ((code = tjsonAddItemToObject(pJson, "nodeInfo", nodeInfo)) < 0) {
tjsonDelete(nodeInfo);
TAOS_CHECK_EXIT(code);
}
for (int32_t i = 0; i < pCfg->totalReplicaNum; ++i) {
SJson *info = tjsonCreateObject();
if (info == NULL) return -1;
if (tjsonAddDoubleToObject(info, "nodePort", pCfg->nodeInfo[i].nodePort) < 0) return -1;
if (tjsonAddStringToObject(info, "nodeFqdn", pCfg->nodeInfo[i].nodeFqdn) < 0) return -1;
if (tjsonAddIntegerToObject(info, "nodeId", pCfg->nodeInfo[i].nodeId) < 0) return -1;
if (tjsonAddIntegerToObject(info, "clusterId", pCfg->nodeInfo[i].clusterId) < 0) return -1;
if (tjsonAddStringToObject(info, "isReplica", syncRoleToStr(pCfg->nodeInfo[i].nodeRole)) < 0) return -1;
if (tjsonAddItemToArray(nodeInfo, info) < 0) return -1;
if (info == NULL) {
TAOS_CHECK_EXIT(TSDB_CODE_OUT_OF_MEMORY);
}
TAOS_CHECK_GOTO(tjsonAddDoubleToObject(info, "nodePort", pCfg->nodeInfo[i].nodePort), NULL, _err);
TAOS_CHECK_GOTO(tjsonAddStringToObject(info, "nodeFqdn", pCfg->nodeInfo[i].nodeFqdn), NULL, _err);
TAOS_CHECK_GOTO(tjsonAddIntegerToObject(info, "nodeId", pCfg->nodeInfo[i].nodeId), NULL, _err);
TAOS_CHECK_GOTO(tjsonAddIntegerToObject(info, "clusterId", pCfg->nodeInfo[i].clusterId), NULL, _err);
TAOS_CHECK_GOTO(tjsonAddStringToObject(info, "isReplica", syncRoleToStr(pCfg->nodeInfo[i].nodeRole)), NULL, _err);
TAOS_CHECK_GOTO(tjsonAddItemToArray(nodeInfo, info), NULL, _err);
continue;
_err:
tjsonDelete(info);
break;
}
return 0;
_exit:
if (code < 0) {
sError("failed to encode sync cfg at line %d since %s", lino, tstrerror(code));
}
TAOS_RETURN(code);
}
static int32_t syncEncodeRaftCfg(const void *pObj, SJson *pJson) {
SRaftCfg *pCfg = (SRaftCfg *)pObj;
if (tjsonAddObject(pJson, "SSyncCfg", syncEncodeSyncCfg, (void *)&pCfg->cfg) < 0) return -1;
if (tjsonAddDoubleToObject(pJson, "isStandBy", pCfg->isStandBy) < 0) return -1;
if (tjsonAddDoubleToObject(pJson, "snapshotStrategy", pCfg->snapshotStrategy) < 0) return -1;
if (tjsonAddDoubleToObject(pJson, "batchSize", pCfg->batchSize) < 0) return -1;
if (tjsonAddIntegerToObject(pJson, "lastConfigIndex", pCfg->lastConfigIndex) < 0) return -1;
if (tjsonAddDoubleToObject(pJson, "configIndexCount", pCfg->configIndexCount) < 0) return -1;
int32_t code = 0, lino = 0;
TAOS_CHECK_EXIT(tjsonAddObject(pJson, "SSyncCfg", syncEncodeSyncCfg, (void *)&pCfg->cfg));
TAOS_CHECK_EXIT(tjsonAddDoubleToObject(pJson, "isStandBy", pCfg->isStandBy));
TAOS_CHECK_EXIT(tjsonAddDoubleToObject(pJson, "snapshotStrategy", pCfg->snapshotStrategy));
TAOS_CHECK_EXIT(tjsonAddDoubleToObject(pJson, "batchSize", pCfg->batchSize));
TAOS_CHECK_EXIT(tjsonAddIntegerToObject(pJson, "lastConfigIndex", pCfg->lastConfigIndex));
TAOS_CHECK_EXIT(tjsonAddDoubleToObject(pJson, "configIndexCount", pCfg->configIndexCount));
SJson *configIndexArr = tjsonCreateArray();
if (configIndexArr == NULL) return -1;
if (tjsonAddItemToObject(pJson, "configIndexArr", configIndexArr) < 0) return -1;
if (configIndexArr == NULL) {
TAOS_CHECK_EXIT(TSDB_CODE_OUT_OF_MEMORY);
}
if ((code = tjsonAddItemToObject(pJson, "configIndexArr", configIndexArr)) < 0) {
tjsonDelete(configIndexArr);
TAOS_CHECK_EXIT(code);
}
for (int32_t i = 0; i < pCfg->configIndexCount; ++i) {
SJson *configIndex = tjsonCreateObject();
if (configIndex == NULL) return -1;
if (tjsonAddIntegerToObject(configIndex, "index", pCfg->configIndexArr[i]) < 0) return -1;
if (tjsonAddItemToArray(configIndexArr, configIndex) < 0) return -1;
if (configIndex == NULL) {
TAOS_CHECK_EXIT(TSDB_CODE_OUT_OF_MEMORY);
}
TAOS_CHECK_EXIT(tjsonAddIntegerToObject(configIndex, "index", pCfg->configIndexArr[i]));
TAOS_CHECK_EXIT(tjsonAddItemToArray(configIndexArr, configIndex));
continue;
_err:
tjsonDelete(configIndex);
break;
}
return 0;
_exit:
if (code < 0) {
sError("failed to encode raft cfg at line %d since %s", lino, tstrerror(code));
}
TAOS_RETURN(code);
}
int32_t syncWriteCfgFile(SSyncNode *pNode) {
int32_t code = -1;
int32_t code = 0, lino = 0;
char *buffer = NULL;
SJson *pJson = NULL;
TdFilePtr pFile = NULL;
const char *realfile = pNode->configPath;
SRaftCfg *pCfg = &pNode->raftCfg;
char file[PATH_MAX] = {0};
snprintf(file, sizeof(file), "%s.bak", realfile);
(void)snprintf(file, sizeof(file), "%s.bak", realfile);
terrno = TSDB_CODE_OUT_OF_MEMORY;
pJson = tjsonCreateObject();
if (pJson == NULL) goto _OVER;
if (tjsonAddObject(pJson, "RaftCfg", syncEncodeRaftCfg, pCfg) < 0) goto _OVER;
if ((pJson = tjsonCreateObject()) == NULL) {
TAOS_CHECK_EXIT(TSDB_CODE_OUT_OF_MEMORY);
}
TAOS_CHECK_EXIT(tjsonAddObject(pJson, "RaftCfg", syncEncodeRaftCfg, pCfg));
buffer = tjsonToString(pJson);
if (buffer == NULL) goto _OVER;
terrno = 0;
if (buffer == NULL) {
TAOS_CHECK_EXIT(TSDB_CODE_OUT_OF_MEMORY);
}
pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
if (pFile == NULL) goto _OVER;
if (pFile == NULL) {
code = terrno ? terrno : TAOS_SYSTEM_ERROR(errno);
TAOS_CHECK_EXIT(code);
}
int32_t len = strlen(buffer);
if (taosWriteFile(pFile, buffer, len) <= 0) goto _OVER;
if (taosFsyncFile(pFile) < 0) goto _OVER;
if (taosWriteFile(pFile, buffer, len) <= 0) {
TAOS_CHECK_EXIT(TAOS_SYSTEM_ERROR(errno));
}
if (taosFsyncFile(pFile) < 0) {
TAOS_CHECK_EXIT(TAOS_SYSTEM_ERROR(errno));
}
taosCloseFile(&pFile);
if (taosRenameFile(file, realfile) != 0) goto _OVER;
(void)taosCloseFile(&pFile);
if (taosRenameFile(file, realfile) != 0) {
TAOS_CHECK_EXIT(TAOS_SYSTEM_ERROR(errno));
}
code = 0;
sInfo("vgId:%d, succeed to write sync cfg file:%s, len:%d, lastConfigIndex:%" PRId64 ", "
"changeVersion:%d", pNode->vgId,
realfile, len, pNode->raftCfg.lastConfigIndex, pNode->raftCfg.cfg.changeVersion);
sInfo("vgId:%d, succeed to write sync cfg file:%s, len:%d, lastConfigIndex:%" PRId64 ", changeVersion:%d",
pNode->vgId, realfile, len, pNode->raftCfg.lastConfigIndex, pNode->raftCfg.cfg.changeVersion);
_OVER:
_exit:
if (pJson != NULL) tjsonDelete(pJson);
if (buffer != NULL) taosMemoryFree(buffer);
if (pFile != NULL) taosCloseFile(&pFile);
if (code != 0) {
if (terrno == 0) terrno = TAOS_SYSTEM_ERROR(errno);
sError("vgId:%d, failed to write sync cfg file:%s since %s", pNode->vgId, realfile, terrstr());
sError("vgId:%d, failed to write sync cfg file:%s since %s", pNode->vgId, realfile, tstrerror(code));
}
return code;
TAOS_RETURN(code);
}
static int32_t syncDecodeSyncCfg(const SJson *pJson, void *pObj) {
@ -135,32 +173,31 @@ static int32_t syncDecodeSyncCfg(const SJson *pJson, void *pObj) {
int32_t code = 0;
tjsonGetInt32ValueFromDouble(pJson, "replicaNum", pCfg->replicaNum, code);
if (code < 0) return -1;
if (code < 0) return TSDB_CODE_INVALID_JSON_FORMAT;
tjsonGetInt32ValueFromDouble(pJson, "myIndex", pCfg->myIndex, code);
if (code < 0) return -1;
if (code < 0) return TSDB_CODE_INVALID_JSON_FORMAT;
tjsonGetInt32ValueFromDouble(pJson, "changeVersion", pCfg->changeVersion, code);
if (code < 0) return -1;
if (code < 0) return TSDB_CODE_INVALID_JSON_FORMAT;
SJson *nodeInfo = tjsonGetObjectItem(pJson, "nodeInfo");
if (nodeInfo == NULL) return -1;
if (nodeInfo == NULL) return TSDB_CODE_INVALID_JSON_FORMAT;
pCfg->totalReplicaNum = tjsonGetArraySize(nodeInfo);
for (int32_t i = 0; i < pCfg->totalReplicaNum; ++i) {
SJson *info = tjsonGetArrayItem(nodeInfo, i);
if (info == NULL) return -1;
if (info == NULL) return TSDB_CODE_INVALID_JSON_FORMAT;
tjsonGetUInt16ValueFromDouble(info, "nodePort", pCfg->nodeInfo[i].nodePort, code);
if (code < 0) return -1;
if (code < 0) return TSDB_CODE_INVALID_JSON_FORMAT;
code = tjsonGetStringValue(info, "nodeFqdn", pCfg->nodeInfo[i].nodeFqdn);
if (code < 0) return -1;
if (code < 0) return TSDB_CODE_INVALID_JSON_FORMAT;
tjsonGetNumberValue(info, "nodeId", pCfg->nodeInfo[i].nodeId, code);
tjsonGetNumberValue(info, "clusterId", pCfg->nodeInfo[i].clusterId, code);
char role[10] = {0};
code = tjsonGetStringValue(info, "isReplica", role);
if(code < 0) return -1;
if(strlen(role) != 0){
if (code < 0) return TSDB_CODE_INVALID_JSON_FORMAT;
if (strlen(role) != 0) {
pCfg->nodeInfo[i].nodeRole = syncStrToRole(role);
}
else{
} else {
pCfg->nodeInfo[i].nodeRole = TAOS_SYNC_ROLE_VOTER;
}
}
@ -172,34 +209,34 @@ static int32_t syncDecodeRaftCfg(const SJson *pJson, void *pObj) {
SRaftCfg *pCfg = (SRaftCfg *)pObj;
int32_t code = 0;
if (tjsonToObject(pJson, "SSyncCfg", syncDecodeSyncCfg, (void *)&pCfg->cfg) < 0) return -1;
TAOS_CHECK_RETURN(tjsonToObject(pJson, "SSyncCfg", syncDecodeSyncCfg, (void *)&pCfg->cfg));
tjsonGetInt8ValueFromDouble(pJson, "isStandBy", pCfg->isStandBy, code);
if (code < 0) return -1;
if (code < 0) return TSDB_CODE_INVALID_JSON_FORMAT;
tjsonGetInt8ValueFromDouble(pJson, "snapshotStrategy", pCfg->snapshotStrategy, code);
if (code < 0) return -1;
if (code < 0) return TSDB_CODE_INVALID_JSON_FORMAT;
tjsonGetInt32ValueFromDouble(pJson, "batchSize", pCfg->batchSize, code);
if (code < 0) return -1;
if (code < 0) return TSDB_CODE_INVALID_JSON_FORMAT;
tjsonGetNumberValue(pJson, "lastConfigIndex", pCfg->lastConfigIndex, code);
if (code < 0) return -1;
if (code < 0) return TSDB_CODE_INVALID_JSON_FORMAT;
tjsonGetInt32ValueFromDouble(pJson, "configIndexCount", pCfg->configIndexCount, code);
if (code < 0) return TSDB_CODE_INVALID_JSON_FORMAT;
SJson *configIndexArr = tjsonGetObjectItem(pJson, "configIndexArr");
if (configIndexArr == NULL) return -1;
if (configIndexArr == NULL) return TSDB_CODE_INVALID_JSON_FORMAT;
pCfg->configIndexCount = tjsonGetArraySize(configIndexArr);
for (int32_t i = 0; i < pCfg->configIndexCount; ++i) {
SJson *configIndex = tjsonGetArrayItem(configIndexArr, i);
if (configIndex == NULL) return -1;
if (configIndex == NULL) return TSDB_CODE_INVALID_JSON_FORMAT;
tjsonGetNumberValue(configIndex, "index", pCfg->configIndexArr[i], code);
if (code < 0) return -1;
if (code < 0) return TSDB_CODE_INVALID_JSON_FORMAT;
}
return 0;
}
int32_t syncReadCfgFile(SSyncNode *pNode) {
int32_t code = -1;
int32_t code = 0;
TdFilePtr pFile = NULL;
char *pData = NULL;
SJson *pJson = NULL;
@ -208,27 +245,27 @@ int32_t syncReadCfgFile(SSyncNode *pNode) {
pFile = taosOpenFile(file, TD_FILE_READ);
if (pFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
sError("vgId:%d, failed to open sync cfg file:%s since %s", pNode->vgId, file, terrstr());
code = TAOS_SYSTEM_ERROR(errno);
sError("vgId:%d, failed to open sync cfg file:%s since %s", pNode->vgId, file, tstrerror(code));
goto _OVER;
}
int64_t size = 0;
if (taosFStatFile(pFile, &size, NULL) < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
sError("vgId:%d, failed to fstat sync cfg file:%s since %s", pNode->vgId, file, terrstr());
code = TAOS_SYSTEM_ERROR(errno);
sError("vgId:%d, failed to fstat sync cfg file:%s since %s", pNode->vgId, file, tstrerror(code));
goto _OVER;
}
pData = taosMemoryMalloc(size + 1);
if (pData == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
code = TSDB_CODE_OUT_OF_MEMORY;
goto _OVER;
}
if (taosReadFile(pFile, pData, size) != size) {
terrno = TAOS_SYSTEM_ERROR(errno);
sError("vgId:%d, failed to read sync cfg file:%s since %s", pNode->vgId, file, terrstr());
code = TAOS_SYSTEM_ERROR(errno);
sError("vgId:%d, failed to read sync cfg file:%s since %s", pNode->vgId, file, tstrerror(code));
goto _OVER;
}
@ -236,18 +273,16 @@ int32_t syncReadCfgFile(SSyncNode *pNode) {
pJson = tjsonParse(pData);
if (pJson == NULL) {
terrno = TSDB_CODE_INVALID_JSON_FORMAT;
code = TSDB_CODE_INVALID_JSON_FORMAT;
goto _OVER;
}
if (tjsonToObject(pJson, "RaftCfg", syncDecodeRaftCfg, (void *)pCfg) < 0) {
terrno = TSDB_CODE_INVALID_JSON_FORMAT;
code = TSDB_CODE_INVALID_JSON_FORMAT;
goto _OVER;
}
code = 0;
sInfo("vgId:%d, succceed to read sync cfg file %s, changeVersion:%d",
pNode->vgId, file, pCfg->cfg.changeVersion);
sInfo("vgId:%d, succceed to read sync cfg file %s, changeVersion:%d", pNode->vgId, file, pCfg->cfg.changeVersion);
_OVER:
if (pData != NULL) taosMemoryFree(pData);
@ -255,15 +290,15 @@ _OVER:
if (pFile != NULL) taosCloseFile(&pFile);
if (code != 0) {
sError("vgId:%d, failed to read sync cfg file:%s since %s", pNode->vgId, file, terrstr());
sError("vgId:%d, failed to read sync cfg file:%s since %s", pNode->vgId, file, tstrerror(code));
}
return code;
TAOS_RETURN(code);
}
int32_t syncAddCfgIndex(SSyncNode *pNode, SyncIndex cfgIndex) {
SRaftCfg *pCfg = &pNode->raftCfg;
if (pCfg->configIndexCount < MAX_CONFIG_INDEX_COUNT) {
return -1;
if (pCfg->configIndexCount >= MAX_CONFIG_INDEX_COUNT) {
return TSDB_CODE_OUT_OF_RANGE;
}
pCfg->configIndexArr[pCfg->configIndexCount] = cfgIndex;

7
source/libs/sync/src/syncRaftEntry.c
View File

@ -99,9 +99,14 @@ void syncEntryDestroy(SSyncRaftEntry* pEntry) {
}
}
void syncEntry2OriginalRpc(const SSyncRaftEntry* pEntry, SRpcMsg* pRpcMsg) {
int32_t syncEntry2OriginalRpc(const SSyncRaftEntry* pEntry, SRpcMsg* pRpcMsg) {
pRpcMsg->msgType = pEntry->originalRpcType;
pRpcMsg->contLen = (int32_t)(pEntry->dataLen);
pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen);
if (pRpcMsg->pCont == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
memcpy(pRpcMsg->pCont, pEntry->data, pRpcMsg->contLen);
return 0;
}

78
source/libs/sync/src/syncRespMgr.c
View File

@ -19,25 +19,33 @@
#include "syncRaftStore.h"
#include "syncUtil.h"
SSyncRespMgr *syncRespMgrCreate(void *data, int64_t ttl) {
SSyncRespMgr *pObj = taosMemoryCalloc(1, sizeof(SSyncRespMgr));
if (pObj == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
int32_t syncRespMgrCreate(void *data, int64_t ttl, SSyncRespMgr **ppObj) {
SSyncRespMgr *pObj = NULL;
*ppObj = NULL;
if ((pObj = taosMemoryCalloc(1, sizeof(SSyncRespMgr))) == NULL) {
TAOS_RETURN(TSDB_CODE_OUT_OF_MEMORY);
}
pObj->pRespHash =
taosHashInit(sizeof(uint64_t), taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
if (pObj->pRespHash == NULL) return NULL;
if (pObj->pRespHash == NULL) {
taosMemoryFree(pObj);
TAOS_RETURN(TSDB_CODE_OUT_OF_MEMORY);
}
pObj->ttl = ttl;
pObj->data = data;
pObj->seqNum = 0;
taosThreadMutexInit(&(pObj->mutex), NULL);
(void)taosThreadMutexInit(&(pObj->mutex), NULL);
SSyncNode *pNode = pObj->data;
sDebug("vgId:%d, resp manager create", pNode->vgId);
return pObj;
*ppObj = pObj;
TAOS_RETURN(0);
}
void syncRespMgrDestroy(SSyncRespMgr *pObj) {
@ -46,82 +54,84 @@ void syncRespMgrDestroy(SSyncRespMgr *pObj) {
SSyncNode *pNode = pObj->data;
sDebug("vgId:%d, resp manager destroy", pNode->vgId);
taosThreadMutexLock(&pObj->mutex);
(void)taosThreadMutexLock(&pObj->mutex);
taosHashCleanup(pObj->pRespHash);
taosThreadMutexUnlock(&pObj->mutex);
taosThreadMutexDestroy(&(pObj->mutex));
(void)taosThreadMutexUnlock(&pObj->mutex);
(void)taosThreadMutexDestroy(&(pObj->mutex));
taosMemoryFree(pObj);
}
uint64_t syncRespMgrAdd(SSyncRespMgr *pObj, const SRespStub *pStub) {
taosThreadMutexLock(&pObj->mutex);
(void)taosThreadMutexLock(&pObj->mutex);
uint64_t seq = ++(pObj->seqNum);
int32_t code = taosHashPut(pObj->pRespHash, &seq, sizeof(uint64_t), pStub, sizeof(SRespStub));
sNTrace(pObj->data, "save message handle:%p, type:%s seq:%" PRIu64 " code:0x%x", pStub->rpcMsg.info.handle,
TMSG_INFO(pStub->rpcMsg.msgType), seq, code);
taosThreadMutexUnlock(&pObj->mutex);
(void)taosThreadMutexUnlock(&pObj->mutex);
return seq;
}
int32_t syncRespMgrDel(SSyncRespMgr *pObj, uint64_t seq) {
taosThreadMutexLock(&pObj->mutex);
(void)taosThreadMutexLock(&pObj->mutex);
int32_t code = taosHashRemove(pObj->pRespHash, &seq, sizeof(seq));
sNTrace(pObj->data, "remove message handle, seq:%" PRIu64 " code:%d", seq, code);
taosThreadMutexUnlock(&pObj->mutex);
(void)taosThreadMutexUnlock(&pObj->mutex);
return code;
}
int32_t syncRespMgrGet(SSyncRespMgr *pObj, uint64_t seq, SRespStub *pStub) {
taosThreadMutexLock(&pObj->mutex);
(void)taosThreadMutexLock(&pObj->mutex);
SRespStub *pTmp = taosHashGet(pObj->pRespHash, &seq, sizeof(uint64_t));
if (pTmp != NULL) {
memcpy(pStub, pTmp, sizeof(SRespStub));
(void)memcpy(pStub, pTmp, sizeof(SRespStub));
sNTrace(pObj->data, "get message handle, type:%s seq:%" PRIu64 " handle:%p", TMSG_INFO(pStub->rpcMsg.msgType), seq,
pStub->rpcMsg.info.handle);
taosThreadMutexUnlock(&pObj->mutex);
(void)taosThreadMutexUnlock(&pObj->mutex);
return 1; // get one object
} else {
sNError(pObj->data, "get message handle, no object of seq:%" PRIu64, seq);
}
taosThreadMutexUnlock(&pObj->mutex);
(void)taosThreadMutexUnlock(&pObj->mutex);
return 0; // get none object
}
int32_t syncRespMgrGetAndDel(SSyncRespMgr *pObj, uint64_t seq, SRpcHandleInfo *pInfo) {
taosThreadMutexLock(&pObj->mutex);
(void)taosThreadMutexLock(&pObj->mutex);
SRespStub *pStub = taosHashGet(pObj->pRespHash, &seq, sizeof(uint64_t));
if (pStub != NULL) {
*pInfo = pStub->rpcMsg.info;
sNTrace(pObj->data, "get-and-del message handle:%p, type:%s seq:%" PRIu64, pStub->rpcMsg.info.handle,
TMSG_INFO(pStub->rpcMsg.msgType), seq);
taosHashRemove(pObj->pRespHash, &seq, sizeof(uint64_t));
(void)taosHashRemove(pObj->pRespHash, &seq, sizeof(uint64_t));
taosThreadMutexUnlock(&pObj->mutex);
(void)taosThreadMutexUnlock(&pObj->mutex);
return 1; // get one object
} else {
sNTrace(pObj->data, "get-and-del message handle, no object of seq:%" PRIu64, seq);
}
taosThreadMutexUnlock(&pObj->mutex);
(void)taosThreadMutexUnlock(&pObj->mutex);
return 0; // get none object
}
static void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl, bool rsp) {
static int32_t syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl, bool rsp) {
SRespStub *pStub = (SRespStub *)taosHashIterate(pObj->pRespHash, NULL);
int cnt = 0;
int sum = 0;
SSyncNode *pNode = pObj->data;
SArray *delIndexArray = taosArrayInit(4, sizeof(uint64_t));
if (delIndexArray == NULL) return;
if (delIndexArray == NULL) {
TAOS_RETURN(TSDB_CODE_OUT_OF_MEMORY);
}
sDebug("vgId:%d, resp manager begin clean by ttl", pNode->vgId);
while (pStub) {
@ -153,7 +163,7 @@ static void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl, bool rsp) {
SRpcMsg rpcMsg = {.info = pStub->rpcMsg.info, .code = TSDB_CODE_SYN_TIMEOUT};
sInfo("vgId:%d, message handle:%p expired, type:%s ahandle:%p", pNode->vgId, rpcMsg.info.handle,
TMSG_INFO(pStub->rpcMsg.msgType), rpcMsg.info.ahandle);
rpcSendResponse(&rpcMsg);
(void)rpcSendResponse(&rpcMsg);
}
pStub = taosHashIterate(pObj->pRespHash, pStub);
@ -164,10 +174,12 @@ static void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl, bool rsp) {
for (int32_t i = 0; i < arraySize; ++i) {
uint64_t *pSeqNum = taosArrayGet(delIndexArray, i);
taosHashRemove(pObj->pRespHash, pSeqNum, sizeof(uint64_t));
(void)taosHashRemove(pObj->pRespHash, pSeqNum, sizeof(uint64_t));
sDebug("vgId:%d, resp manager clean by ttl, seq:%" PRId64, pNode->vgId, *pSeqNum);
}
taosArrayDestroy(delIndexArray);
return 0;
}
void syncRespCleanRsp(SSyncRespMgr *pObj) {
@ -176,16 +188,16 @@ void syncRespCleanRsp(SSyncRespMgr *pObj) {
SSyncNode *pNode = pObj->data;
sTrace("vgId:%d, clean all resp", pNode->vgId);
taosThreadMutexLock(&pObj->mutex);
syncRespCleanByTTL(pObj, -1, true);
taosThreadMutexUnlock(&pObj->mutex);
(void)taosThreadMutexLock(&pObj->mutex);
(void)syncRespCleanByTTL(pObj, -1, true);
(void)taosThreadMutexUnlock(&pObj->mutex);
}
void syncRespClean(SSyncRespMgr *pObj) {
SSyncNode *pNode = pObj->data;
sTrace("vgId:%d, clean resp by ttl", pNode->vgId);
taosThreadMutexLock(&pObj->mutex);
syncRespCleanByTTL(pObj, pObj->ttl, false);
taosThreadMutexUnlock(&pObj->mutex);
(void)taosThreadMutexLock(&pObj->mutex);
(void)syncRespCleanByTTL(pObj, pObj->ttl, false);
(void)taosThreadMutexUnlock(&pObj->mutex);
}

392
source/libs/sync/src/syncSnapshot.c
View File

File diff suppressed because it is too large Load Diff

8
source/libs/sync/src/syncTimeout.c
View File

@ -65,7 +65,7 @@ static int32_t syncNodeTimerRoutine(SSyncNode* ths) {
}
// timer replicate
syncNodeReplicate(ths);
(void)syncNodeReplicate(ths);
// clean mnode index
if (syncNodeIsMnode(ths)) {
@ -89,7 +89,7 @@ static int32_t syncNodeTimerRoutine(SSyncNode* ths) {
snapshotSenderStop(pSender, false);
} else {
sSWarn(pSender, "snap replication resend.");
snapshotReSend(pSender);
(void)snapshotReSend(pSender);
}
}
}
@ -112,14 +112,14 @@ int32_t syncNodeOnTimeout(SSyncNode* ths, const SRpcMsg* pRpc) {
if (atomic_load_64(&ths->pingTimerLogicClockUser) <= pMsg->logicClock) {
++(ths->pingTimerCounter);
syncNodeTimerRoutine(ths);
(void)syncNodeTimerRoutine(ths);
}
} else if (pMsg->timeoutType == SYNC_TIMEOUT_ELECTION) {
if (atomic_load_64(&ths->electTimerLogicClock) <= pMsg->logicClock) {
++(ths->electTimerCounter);
syncNodeElect(ths);
(void)syncNodeElect(ths);
}
} else if (pMsg->timeoutType == SYNC_TIMEOUT_HEARTBEAT) {

4
source/libs/sync/src/syncUtil.c
View File

@ -103,10 +103,10 @@ bool syncUtilUserPreCommit(tmsg_t msgType) { return msgType != TDMT_SYNC_NOOP &&
bool syncUtilUserRollback(tmsg_t msgType) { return msgType != TDMT_SYNC_NOOP && msgType != TDMT_SYNC_LEADER_TRANSFER; }
void syncUtilGenerateArbToken(int32_t nodeId, int32_t groupId, char* buf) {
memset(buf, 0, TSDB_ARB_TOKEN_SIZE);
(void)memset(buf, 0, TSDB_ARB_TOKEN_SIZE);
int32_t randVal = taosSafeRand() % 1000;
int64_t currentMs = taosGetTimestampMs();
snprintf(buf, TSDB_ARB_TOKEN_SIZE, "d%d#g%d#%" PRId64 "#%d", nodeId, groupId, currentMs, randVal);
(void)snprintf(buf, TSDB_ARB_TOKEN_SIZE, "d%d#g%d#%" PRId64 "#%d", nodeId, groupId, currentMs, randVal);
}
// for leader

4
source/libs/sync/src/syncVoteMgr.c
View File

@ -19,7 +19,7 @@
#include "syncUtil.h"
static void voteGrantedClearVotes(SVotesGranted *pVotesGranted) {
memset(pVotesGranted->isGranted, 0, sizeof(pVotesGranted->isGranted));
(void)memset(pVotesGranted->isGranted, 0, sizeof(pVotesGranted->isGranted));
pVotesGranted->votes = 0;
}
@ -165,5 +165,5 @@ void votesRespondAdd(SVotesRespond *pVotesRespond, const SyncRequestVoteReply *p
void votesRespondReset(SVotesRespond *pVotesRespond, SyncTerm term) {
pVotesRespond->term = term;
memset(pVotesRespond->isRespond, 0, sizeof(pVotesRespond->isRespond));
(void)memset(pVotesRespond->isRespond, 0, sizeof(pVotesRespond->isRespond));
}

8
source/libs/sync/test/syncRespMgrTest.cpp
View File

@ -77,7 +77,7 @@ SSyncNode *createSyncNode() {
void test1() {
printf("------- test1 ---------\n");
pMgr = syncRespMgrCreate(createSyncNode(), 0);
(void)syncRespMgrCreate(createSyncNode(), 0, &pMgr);
assert(pMgr != NULL);
syncRespMgrInsert(10);
@ -102,7 +102,7 @@ void test1() {
void test2() {
printf("------- test2 ---------\n");
pMgr = syncRespMgrCreate(createSyncNode(), 0);
(void)syncRespMgrCreate(createSyncNode(), 0, &pMgr);
assert(pMgr != NULL);
syncRespMgrInsert(10);
@ -119,7 +119,7 @@ void test2() {
void test3() {
printf("------- test3 ---------\n");
pMgr = syncRespMgrCreate(createSyncNode(), 0);
(void)syncRespMgrCreate(createSyncNode(), 0, &pMgr);
assert(pMgr != NULL);
syncRespMgrInsert(10);
@ -136,7 +136,7 @@ void test3() {
void test4() {
printf("------- test4 ---------\n");
pMgr = syncRespMgrCreate(createSyncNode(), 2);
(void)syncRespMgrCreate(createSyncNode(), 2, &pMgr);
assert(pMgr != NULL);
syncRespMgrInsert(5);

3
source/libs/sync/test/syncSnapshotReceiverTest.cpp
View File

@ -42,7 +42,8 @@ SSyncSnapshotReceiver* createReceiver() {
id.addr = syncUtilAddr2U64("1.2.3.4", 99);
id.vgId = 100;
SSyncSnapshotReceiver* pReceiver = snapshotReceiverCreate(pSyncNode, id);
SSyncSnapshotReceiver* pReceiver = NULL;
(void)snapshotReceiverCreate(pSyncNode, id, &pReceiver);
pReceiver->start = true;
pReceiver->ack = 20;
pReceiver->pWriter = (void*)0x11;

3
source/libs/sync/test/syncSnapshotSenderTest.cpp
View File

@ -39,7 +39,8 @@ SSyncSnapshotSender* createSender() {
pSyncNode->pFsm->FpGetSnapshotInfo = GetSnapshot;
#endif
SSyncSnapshotSender* pSender = snapshotSenderCreate(pSyncNode, 2);
SSyncSnapshotSender* pSender = NULL;
(void)snapshotSenderCreate(pSyncNode, 2, &pSender);
pSender->start = true;
pSender->seq = 10;
pSender->ack = 20;

48
source/libs/tfs/src/tfs.c
View File

@ -81,7 +81,7 @@ void tfsClose(STfs *pTfs) {
}
taosHashCleanup(pTfs->hash);
taosThreadSpinDestroy(&pTfs->lock);
(void)taosThreadSpinDestroy(&pTfs->lock);
taosMemoryFree(pTfs);
}
@ -96,15 +96,15 @@ void tfsUpdateSize(STfs *pTfs) {
size.used += pTier->size.used;
}
tfsLock(pTfs);
(void)tfsLock(pTfs);
pTfs->size = size;
tfsUnLock(pTfs);
(void)tfsUnLock(pTfs);
}
SDiskSize tfsGetSize(STfs *pTfs) {
tfsLock(pTfs);
(void)tfsLock(pTfs);
SDiskSize size = pTfs->size;
tfsUnLock(pTfs);
(void)tfsUnLock(pTfs);
return size;
}
@ -190,7 +190,7 @@ void tfsInitFile(STfs *pTfs, STfsFile *pFile, SDiskID diskId, const char *rname)
tstrncpy(pFile->rname, rname, TSDB_FILENAME_LEN);
char tmpName[TMPNAME_LEN] = {0};
snprintf(tmpName, TMPNAME_LEN, "%s%s%s", pDisk->path, TD_DIRSEP, rname);
(void)snprintf(tmpName, TMPNAME_LEN, "%s%s%s", pDisk->path, TD_DIRSEP, rname);
tstrncpy(pFile->aname, tmpName, TSDB_FILENAME_LEN);
pFile->pTfs = pTfs;
}
@ -200,12 +200,12 @@ bool tfsIsSameFile(const STfsFile *pFile1, const STfsFile *pFile2) {
if (pFile1->did.level != pFile2->did.level) return false;
if (pFile1->did.id != pFile2->did.id) return false;
char nameBuf1[TMPNAME_LEN], nameBuf2[TMPNAME_LEN];
strncpy(nameBuf1, pFile1->rname, TMPNAME_LEN);
strncpy(nameBuf2, pFile2->rname, TMPNAME_LEN);
(void)strncpy(nameBuf1, pFile1->rname, TMPNAME_LEN);
(void)strncpy(nameBuf2, pFile2->rname, TMPNAME_LEN);
nameBuf1[TMPNAME_LEN - 1] = 0;
nameBuf2[TMPNAME_LEN - 1] = 0;
taosRealPath(nameBuf1, NULL, TMPNAME_LEN);
taosRealPath(nameBuf2, NULL, TMPNAME_LEN);
(void)taosRealPath(nameBuf1, NULL, TMPNAME_LEN);
(void)taosRealPath(nameBuf2, NULL, TMPNAME_LEN);
if (strncmp(nameBuf1, nameBuf2, TMPNAME_LEN) != 0) return false;
return true;
}
@ -251,7 +251,7 @@ void tfsDirname(const STfsFile *pFile, char *dest) {
void tfsAbsoluteName(STfs *pTfs, SDiskID diskId, const char *rname, char *aname) {
STfsDisk *pDisk = TFS_DISK_AT(pTfs, diskId);
snprintf(aname, TSDB_FILENAME_LEN, "%s%s%s", pDisk->path, TD_DIRSEP, rname);
(void)snprintf(aname, TSDB_FILENAME_LEN, "%s%s%s", pDisk->path, TD_DIRSEP, rname);
}
int32_t tfsRemoveFile(const STfsFile *pFile) { return taosRemoveFile(pFile->aname); }
@ -267,7 +267,7 @@ int32_t tfsMkdirAt(STfs *pTfs, const char *rname, SDiskID diskId) {
if (pDisk == NULL) {
TAOS_RETURN(TSDB_CODE_FS_INVLD_CFG);
}
snprintf(aname, TMPNAME_LEN, "%s%s%s", pDisk->path, TD_DIRSEP, rname);
(void)snprintf(aname, TMPNAME_LEN, "%s%s%s", pDisk->path, TD_DIRSEP, rname);
if (taosMkDir(aname) != 0) {
TAOS_RETURN(TAOS_SYSTEM_ERROR(errno));
}
@ -347,7 +347,7 @@ bool tfsDirExistAt(STfs *pTfs, const char *rname, SDiskID diskId) {
STfsDisk *pDisk = TFS_DISK_AT(pTfs, diskId);
char aname[TMPNAME_LEN];
snprintf(aname, TMPNAME_LEN, "%s%s%s", pDisk->path, TD_DIRSEP, rname);
(void)snprintf(aname, TMPNAME_LEN, "%s%s%s", pDisk->path, TD_DIRSEP, rname);
return taosDirExist(aname);
}
@ -362,7 +362,7 @@ int32_t tfsRmdir(STfs *pTfs, const char *rname) {
STfsTier *pTier = TFS_TIER_AT(pTfs, level);
for (int32_t id = 0; id < pTier->ndisk; id++) {
STfsDisk *pDisk = pTier->disks[id];
snprintf(aname, TMPNAME_LEN, "%s%s%s", pDisk->path, TD_DIRSEP, rname);
(void)snprintf(aname, TMPNAME_LEN, "%s%s%s", pDisk->path, TD_DIRSEP, rname);
uInfo("tfs remove dir:%s aname:%s rname:[%s]", pDisk->path, aname, rname);
taosRemoveDir(aname);
}
@ -379,8 +379,8 @@ static int32_t tfsRenameAt(STfs *pTfs, SDiskID diskId, const char *orname, const
int32_t id = diskId.id;
STfsTier *pTier = TFS_TIER_AT(pTfs, level);
STfsDisk *pDisk = pTier->disks[id];
snprintf(oaname, TMPNAME_LEN, "%s%s%s", pDisk->path, TD_DIRSEP, orname);
snprintf(naname, TMPNAME_LEN, "%s%s%s", pDisk->path, TD_DIRSEP, nrname);
(void)snprintf(oaname, TMPNAME_LEN, "%s%s%s", pDisk->path, TD_DIRSEP, orname);
(void)snprintf(naname, TMPNAME_LEN, "%s%s%s", pDisk->path, TD_DIRSEP, nrname);
if (taosRenameFile(oaname, naname) != 0 && errno != ENOENT) {
int32_t code = TAOS_SYSTEM_ERROR(errno); // TODO: use return value of taosRenameFile directly
@ -417,7 +417,7 @@ int32_t tfsSearch(STfs *pTfs, int32_t level, const char *fname) {
for (int32_t id = 0; id < pTier->ndisk; id++) {
STfsDisk *pDisk = pTier->disks[id];
snprintf(path, TMPNAME_LEN - 1, "%s%s%s", pDisk->path, TD_DIRSEP, fname);
(void)snprintf(path, TMPNAME_LEN - 1, "%s%s%s", pDisk->path, TD_DIRSEP, fname);
if (taosCheckExistFile(path)) {
return id;
}
@ -460,9 +460,9 @@ const STfsFile *tfsReaddir(STfsDir *pTfsDir) {
if (strcmp(name, ".") == 0 || strcmp(name, "..") == 0) continue;
if (pTfsDir->dirName[0] == 0) {
snprintf(bname, TMPNAME_LEN * 2, "%s", name);
(void)snprintf(bname, TMPNAME_LEN * 2, "%s", name);
} else {
snprintf(bname, TMPNAME_LEN * 2, "%s%s%s", pTfsDir->dirName, TD_DIRSEP, name);
(void)snprintf(bname, TMPNAME_LEN * 2, "%s%s%s", pTfsDir->dirName, TD_DIRSEP, name);
}
tfsInitFile(pTfsDir->pTfs, &pTfsDir->tfile, pTfsDir->did, bname);
@ -655,9 +655,9 @@ static int32_t tfsOpendirImpl(STfs *pTfs, STfsDir *pTfsDir) {
pTfsDir->did.id = pDisk->id;
if (pDisk->path == NULL || pDisk->path[0] == 0) {
snprintf(adir, TMPNAME_LEN * 2, "%s", pTfsDir->dirName);
(void)snprintf(adir, TMPNAME_LEN * 2, "%s", pTfsDir->dirName);
} else {
snprintf(adir, TMPNAME_LEN * 2, "%s%s%s", pDisk->path, TD_DIRSEP, pTfsDir->dirName);
(void)snprintf(adir, TMPNAME_LEN * 2, "%s%s%s", pDisk->path, TD_DIRSEP, pTfsDir->dirName);
}
pTfsDir->pDir = taosOpenDir(adir);
if (pTfsDir->pDir != NULL) break;
@ -698,7 +698,7 @@ int32_t tfsGetMonitorInfo(STfs *pTfs, SMonDiskInfo *pInfo) {
tfsUpdateSize(pTfs);
tfsLock(pTfs);
(void)tfsLock(pTfs);
for (int32_t level = 0; level < pTfs->nlevel; level++) {
STfsTier *pTier = &pTfs->tiers[level];
for (int32_t disk = 0; disk < pTier->ndisk; ++disk) {
@ -708,14 +708,14 @@ int32_t tfsGetMonitorInfo(STfs *pTfs, SMonDiskInfo *pInfo) {
dinfo.level = pDisk->level;
tstrncpy(dinfo.name, pDisk->path, sizeof(dinfo.name));
if (taosArrayPush(pInfo->datadirs, &dinfo) == NULL) {
tfsUnLock(pTfs);
(void)tfsUnLock(pTfs);
taosArrayDestroy(pInfo->datadirs);
pInfo->datadirs = NULL;
TAOS_RETURN(TSDB_CODE_OUT_OF_MEMORY);
}
}
}
tfsUnLock(pTfs);
(void)tfsUnLock(pTfs);
TAOS_RETURN(0);
}

14
source/libs/tfs/src/tfsTier.c
View File

@ -19,7 +19,7 @@
extern int64_t tsMinDiskFreeSize;
int32_t tfsInitTier(STfsTier *pTier, int32_t level) {
memset(pTier, 0, sizeof(STfsTier));
(void)memset(pTier, 0, sizeof(STfsTier));
if (taosThreadSpinInit(&pTier->lock, 0) != 0) {
TAOS_RETURN(TAOS_SYSTEM_ERROR(errno));
@ -35,7 +35,7 @@ void tfsDestroyTier(STfsTier *pTier) {
}
pTier->ndisk = 0;
taosThreadSpinDestroy(&pTier->lock);
(void)taosThreadSpinDestroy(&pTier->lock);
}
int32_t tfsMountDiskToTier(STfsTier *pTier, SDiskCfg *pCfg, STfsDisk **ppDisk) {
@ -88,7 +88,7 @@ void tfsUpdateTierSize(STfsTier *pTier) {
SDiskSize size = {0};
int32_t nAvailDisks = 0;
tfsLockTier(pTier);
(void)tfsLockTier(pTier);
for (int32_t id = 0; id < pTier->ndisk; id++) {
STfsDisk *pDisk = pTier->disks[id];
@ -104,15 +104,15 @@ void tfsUpdateTierSize(STfsTier *pTier) {
pTier->size = size;
pTier->nAvailDisks = nAvailDisks;
tfsUnLockTier(pTier);
(void)tfsUnLockTier(pTier);
}
// Round-Robin to allocate disk on a tier
int32_t tfsAllocDiskOnTier(STfsTier *pTier) {
tfsLockTier(pTier);
(void)tfsLockTier(pTier);
if (pTier->ndisk <= 0 || pTier->nAvailDisks <= 0) {
tfsUnLockTier(pTier);
(void)tfsUnLockTier(pTier);
TAOS_RETURN(TSDB_CODE_FS_NO_VALID_DISK);
}
@ -155,7 +155,7 @@ int32_t tfsAllocDiskOnTier(STfsTier *pTier) {
#endif
}
tfsUnLockTier(pTier);
(void)tfsUnLockTier(pTier);
if (retId < 0) {
TAOS_RETURN(TSDB_CODE_FS_NO_VALID_DISK);
}

1
source/os/src/osFile.c
View File

@ -1158,6 +1158,7 @@ TdFilePtr taosOpenFile(const char *path, int32_t tdFileOptions) {
if (fd >= 0) (void)close(fd);
#endif
if (fp != NULL) (void)fclose(fp);
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}

4
source/util/src/tbase58.c
View File

@ -67,7 +67,7 @@ int32_t base58_encode(const uint8_t *value, int32_t vlen, char **result) {
if (bfree) taosMemoryFree(pbuf);
return TSDB_CODE_OUT_OF_MEMORY;
}
memset(pResult, '1', nz);
(void)memset(pResult, '1', nz);
while (pi != pbuf + size) pResult[nz++] = basis_58[*pi++];
if (bfree) taosMemoryFree(pbuf);
@ -152,7 +152,7 @@ int32_t base58_decode(const char *value, size_t inlen, int32_t *outlen, uint8_t
return TSDB_CODE_OUT_OF_MEMORY;
}
memset(pResult, 0, nz);
(void)memset(pResult, 0, nz);
while (it != pbuf + size) pResult[nz++] = *it++;
if (outlen) *outlen = nz;