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

Merge branch '3.0' of https://github.com/taosdata/TDengine into enh/-TD-31890-15

This commit is contained in:
Hongze Cheng 2024-09-24 13:20:20 +08:00
parent 3c4b16b8bc a7a5c44666
commit 4c23f9d5bd
15 changed files with 622 additions and 219 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
Jenkinsfile2
View File

@ -306,9 +306,9 @@ def pre_test_build_win() {
cd %WIN_CONNECTOR_ROOT% cd %WIN_CONNECTOR_ROOT%
python.exe -m pip install --upgrade pip python.exe -m pip install --upgrade pip
python -m pip uninstall taospy -y python -m pip uninstall taospy -y
python -m pip install taospy==2.7.13 python -m pip install taospy==2.7.16
python -m pip uninstall taos-ws-py -y python -m pip uninstall taos-ws-py -y
python -m pip install taos-ws-py==0.3.1 python -m pip install taos-ws-py==0.3.3
xcopy /e/y/i/f %WIN_INTERNAL_ROOT%\\debug\\build\\lib\\taos.dll C:\\Windows\\System32 xcopy /e/y/i/f %WIN_INTERNAL_ROOT%\\debug\\build\\lib\\taos.dll C:\\Windows\\System32
''' '''
return 1 return 1

8
source/dnode/mnode/impl/src/mndSync.c
View File

@ -497,11 +497,9 @@ int32_t mndInitSync(SMnode *pMnode) {
pNode->nodePort = pMgmt->replicas[i].port; pNode->nodePort = pMgmt->replicas[i].port;
tstrncpy(pNode->nodeFqdn, pMgmt->replicas[i].fqdn, sizeof(pNode->nodeFqdn)); tstrncpy(pNode->nodeFqdn, pMgmt->replicas[i].fqdn, sizeof(pNode->nodeFqdn));
pNode->nodeRole = pMgmt->nodeRoles[i]; pNode->nodeRole = pMgmt->nodeRoles[i];
if (tmsgUpdateDnodeInfo(&pNode->nodeId, &pNode->clusterId, pNode->nodeFqdn, &pNode->nodePort) != true) { bool update = tmsgUpdateDnodeInfo(&pNode->nodeId, &pNode->clusterId, pNode->nodeFqdn, &pNode->nodePort);
mError("failed to open sync, tmsgUpdateDnodeInfo is false"); mInfo("vgId:1, index:%d ep:%s:%u dnode:%d cluster:%" PRId64 ", update:%d", i, pNode->nodeFqdn, pNode->nodePort,
} pNode->nodeId, pNode->clusterId, update);
mInfo("vgId:1, index:%d ep:%s:%u dnode:%d cluster:%" PRId64, i, pNode->nodeFqdn, pNode->nodePort, pNode->nodeId,
pNode->clusterId);
} }
int32_t code = 0; int32_t code = 0;

302
source/dnode/vnode/src/tsdb/tsdbCache.c
View File

@ -577,7 +577,10 @@ static void tsdbCacheDeleter(const void *key, size_t klen, void *value, void *ud
SLastCol *pLastCol = (SLastCol *)value; SLastCol *pLastCol = (SLastCol *)value;
if (pLastCol->dirty) { if (pLastCol->dirty) {
(void)tsdbCacheFlushDirty(key, klen, pLastCol, ud); if (tsdbCacheFlushDirty(key, klen, pLastCol, ud) != 0) {
STsdb *pTsdb = (STsdb *)ud;
tsdbError("tsdb/cache: vgId:%d, flush cache %s failed at line %d.", TD_VID(pTsdb->pVnode), __func__, __LINE__);
}
} }
for (uint8_t i = 0; i < pLastCol->rowKey.numOfPKs; ++i) { for (uint8_t i = 0; i < pLastCol->rowKey.numOfPKs; ++i) {
@ -719,14 +722,22 @@ static int32_t tsdbCacheDropTableColumn(STsdb *pTsdb, int64_t uid, int16_t cid,
rocksdb_writebatch_t *wb = pTsdb->rCache.writebatch; rocksdb_writebatch_t *wb = pTsdb->rCache.writebatch;
{ {
SLastCol *pLastCol = NULL; SLastCol *pLastCol = NULL;
(void)tsdbCacheDeserialize(values_list[0], values_list_sizes[0], &pLastCol); code = tsdbCacheDeserialize(values_list[0], values_list_sizes[0], &pLastCol);
if (code != TSDB_CODE_SUCCESS) {
tsdbWarn("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
}
if (NULL != pLastCol) { if (NULL != pLastCol) {
rocksdb_writebatch_delete(wb, keys_list[0], klen); rocksdb_writebatch_delete(wb, keys_list[0], klen);
} }
taosMemoryFreeClear(pLastCol); taosMemoryFreeClear(pLastCol);
pLastCol = NULL; pLastCol = NULL;
(void)tsdbCacheDeserialize(values_list[1], values_list_sizes[1], &pLastCol); code = tsdbCacheDeserialize(values_list[1], values_list_sizes[1], &pLastCol);
if (code != TSDB_CODE_SUCCESS) {
tsdbWarn("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
}
if (NULL != pLastCol) { if (NULL != pLastCol) {
rocksdb_writebatch_delete(wb, keys_list[1], klen); rocksdb_writebatch_delete(wb, keys_list[1], klen);
} }
@ -738,7 +749,9 @@ static int32_t tsdbCacheDropTableColumn(STsdb *pTsdb, int64_t uid, int16_t cid,
for (int i = 0; i < 2; i++) { for (int i = 0; i < 2; i++) {
LRUHandle *h = taosLRUCacheLookup(pTsdb->lruCache, keys_list[i], klen); LRUHandle *h = taosLRUCacheLookup(pTsdb->lruCache, keys_list[i], klen);
if (h) { if (h) {
(void)taosLRUCacheRelease(pTsdb->lruCache, h, true); if (taosLRUCacheRelease(pTsdb->lruCache, h, true)) {
tsdbInfo("vgId:%d, %s release lru cache failed at line %d.", TD_VID(pTsdb->pVnode), __func__, __LINE__);
}
taosLRUCacheErase(pTsdb->lruCache, keys_list[i], klen); taosLRUCacheErase(pTsdb->lruCache, keys_list[i], klen);
} }
} }
@ -765,8 +778,20 @@ int32_t tsdbCacheNewTable(STsdb *pTsdb, tb_uid_t uid, tb_uid_t suid, SSchemaWrap
int16_t cid = pSchemaRow->pSchema[i].colId; int16_t cid = pSchemaRow->pSchema[i].colId;
int8_t col_type = pSchemaRow->pSchema[i].type; int8_t col_type = pSchemaRow->pSchema[i].type;
(void)tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, LFLAG_LAST_ROW); code = tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, LFLAG_LAST_ROW);
(void)tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, LFLAG_LAST); if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s new table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
}
code = tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, LFLAG_LAST);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s new table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
}
} }
} else { } else {
STSchema *pTSchema = NULL; STSchema *pTSchema = NULL;
@ -781,8 +806,20 @@ int32_t tsdbCacheNewTable(STsdb *pTsdb, tb_uid_t uid, tb_uid_t suid, SSchemaWrap
int16_t cid = pTSchema->columns[i].colId; int16_t cid = pTSchema->columns[i].colId;
int8_t col_type = pTSchema->columns[i].type; int8_t col_type = pTSchema->columns[i].type;
(void)tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, LFLAG_LAST_ROW); code = tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, LFLAG_LAST_ROW);
(void)tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, LFLAG_LAST); if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s new table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
}
code = tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, LFLAG_LAST);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s new table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
}
} }
taosMemoryFree(pTSchema); taosMemoryFree(pTSchema);
@ -798,7 +835,13 @@ int32_t tsdbCacheDropTable(STsdb *pTsdb, tb_uid_t uid, tb_uid_t suid, SSchemaWra
(void)taosThreadMutexLock(&pTsdb->lruMutex); (void)taosThreadMutexLock(&pTsdb->lruMutex);
(void)tsdbCacheCommitNoLock(pTsdb); code = tsdbCacheCommitNoLock(pTsdb);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s commit with no lock failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
}
if (pSchemaRow != NULL) { if (pSchemaRow != NULL) {
bool hasPrimayKey = false; bool hasPrimayKey = false;
@ -810,7 +853,13 @@ int32_t tsdbCacheDropTable(STsdb *pTsdb, tb_uid_t uid, tb_uid_t suid, SSchemaWra
int16_t cid = pSchemaRow->pSchema[i].colId; int16_t cid = pSchemaRow->pSchema[i].colId;
int8_t col_type = pSchemaRow->pSchema[i].type; int8_t col_type = pSchemaRow->pSchema[i].type;
(void)tsdbCacheDropTableColumn(pTsdb, uid, cid, hasPrimayKey); code = tsdbCacheDropTableColumn(pTsdb, uid, cid, hasPrimayKey);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s drop table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
}
} }
} else { } else {
STSchema *pTSchema = NULL; STSchema *pTSchema = NULL;
@ -830,7 +879,13 @@ int32_t tsdbCacheDropTable(STsdb *pTsdb, tb_uid_t uid, tb_uid_t suid, SSchemaWra
int16_t cid = pTSchema->columns[i].colId; int16_t cid = pTSchema->columns[i].colId;
int8_t col_type = pTSchema->columns[i].type; int8_t col_type = pTSchema->columns[i].type;
(void)tsdbCacheDropTableColumn(pTsdb, uid, cid, hasPrimayKey); code = tsdbCacheDropTableColumn(pTsdb, uid, cid, hasPrimayKey);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s drop table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
}
} }
taosMemoryFree(pTSchema); taosMemoryFree(pTSchema);
@ -848,7 +903,13 @@ int32_t tsdbCacheDropSubTables(STsdb *pTsdb, SArray *uids, tb_uid_t suid) {
(void)taosThreadMutexLock(&pTsdb->lruMutex); (void)taosThreadMutexLock(&pTsdb->lruMutex);
(void)tsdbCacheCommitNoLock(pTsdb); code = tsdbCacheCommitNoLock(pTsdb);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s commit with no lock failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
}
STSchema *pTSchema = NULL; STSchema *pTSchema = NULL;
code = metaGetTbTSchemaEx(pTsdb->pVnode->pMeta, suid, suid, -1, &pTSchema); code = metaGetTbTSchemaEx(pTsdb->pVnode->pMeta, suid, suid, -1, &pTSchema);
@ -871,7 +932,14 @@ int32_t tsdbCacheDropSubTables(STsdb *pTsdb, SArray *uids, tb_uid_t suid) {
int16_t cid = pTSchema->columns[i].colId; int16_t cid = pTSchema->columns[i].colId;
int8_t col_type = pTSchema->columns[i].type; int8_t col_type = pTSchema->columns[i].type;
(void)tsdbCacheDropTableColumn(pTsdb, uid, cid, hasPrimayKey); code = tsdbCacheDropTableColumn(pTsdb, uid, cid, hasPrimayKey);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s drop table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
taosMemoryFree(pTSchema);
TAOS_RETURN(code);
}
} }
} }
@ -889,12 +957,22 @@ int32_t tsdbCacheNewNTableColumn(STsdb *pTsdb, int64_t uid, int16_t cid, int8_t
(void)taosThreadMutexLock(&pTsdb->lruMutex); (void)taosThreadMutexLock(&pTsdb->lruMutex);
(void)tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, 0); code = tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, 0);
(void)tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, 1); if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s new table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
}
code = tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, 1);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s new table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
}
// rocksMayWrite(pTsdb, true, false, false); // rocksMayWrite(pTsdb, true, false, false);
(void)taosThreadMutexUnlock(&pTsdb->lruMutex); (void)taosThreadMutexUnlock(&pTsdb->lruMutex);
//(void)tsdbCacheCommit(pTsdb);
TAOS_RETURN(code); TAOS_RETURN(code);
} }
@ -904,9 +982,21 @@ int32_t tsdbCacheDropNTableColumn(STsdb *pTsdb, int64_t uid, int16_t cid, bool h
(void)taosThreadMutexLock(&pTsdb->lruMutex); (void)taosThreadMutexLock(&pTsdb->lruMutex);
(void)tsdbCacheCommitNoLock(pTsdb); code = tsdbCacheCommitNoLock(pTsdb);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s commit with no lock failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
}
(void)tsdbCacheDropTableColumn(pTsdb, uid, cid, hasPrimayKey); code = tsdbCacheDropTableColumn(pTsdb, uid, cid, hasPrimayKey);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s drop table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
}
rocksMayWrite(pTsdb, false); rocksMayWrite(pTsdb, false);
@ -923,14 +1013,24 @@ int32_t tsdbCacheNewSTableColumn(STsdb *pTsdb, SArray *uids, int16_t cid, int8_t
for (int i = 0; i < TARRAY_SIZE(uids); ++i) { for (int i = 0; i < TARRAY_SIZE(uids); ++i) {
tb_uid_t uid = ((tb_uid_t *)TARRAY_DATA(uids))[i]; tb_uid_t uid = ((tb_uid_t *)TARRAY_DATA(uids))[i];
(void)tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, 0); code = tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, 0);
(void)tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, 1); if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s new table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
}
code = tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, 1);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s new table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
}
} }
// rocksMayWrite(pTsdb, true, false, false); // rocksMayWrite(pTsdb, true, false, false);
(void)taosThreadMutexUnlock(&pTsdb->lruMutex); (void)taosThreadMutexUnlock(&pTsdb->lruMutex);
//(void)tsdbCacheCommit(pTsdb);
TAOS_RETURN(code); TAOS_RETURN(code);
} }
@ -939,12 +1039,24 @@ int32_t tsdbCacheDropSTableColumn(STsdb *pTsdb, SArray *uids, int16_t cid, bool
(void)taosThreadMutexLock(&pTsdb->lruMutex); (void)taosThreadMutexLock(&pTsdb->lruMutex);
(void)tsdbCacheCommitNoLock(pTsdb); code = tsdbCacheCommitNoLock(pTsdb);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s commit with no lock failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
}
for (int i = 0; i < TARRAY_SIZE(uids); ++i) { for (int i = 0; i < TARRAY_SIZE(uids); ++i) {
int64_t uid = ((tb_uid_t *)TARRAY_DATA(uids))[i]; int64_t uid = ((tb_uid_t *)TARRAY_DATA(uids))[i];
(void)tsdbCacheDropTableColumn(pTsdb, uid, cid, hasPrimayKey); code = tsdbCacheDropTableColumn(pTsdb, uid, cid, hasPrimayKey);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s drop table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
}
} }
rocksMayWrite(pTsdb, false); rocksMayWrite(pTsdb, false);
@ -1109,7 +1221,9 @@ static int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, SArray
} }
} }
(void)taosLRUCacheRelease(pCache, h, false); if (!taosLRUCacheRelease(pCache, h, false)) {
tsdbInfo("vgId:%d, %s release lru cache failed at line %d", TD_VID(pTsdb->pVnode), __func__, __LINE__);
}
TAOS_CHECK_EXIT(code); TAOS_CHECK_EXIT(code);
} else { } else {
if (!remainCols) { if (!remainCols) {
@ -1151,7 +1265,7 @@ static int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, SArray
keys_list_sizes[i] = ROCKS_KEY_LEN; keys_list_sizes[i] = ROCKS_KEY_LEN;
} }
rocksMayWrite(pTsdb, true); // flush writebatch cache rocksMayWrite(pTsdb, true); // flush writebatch cache
code = tsdbCacheGetValuesFromRocks(pTsdb, num_keys, (const char *const *)keys_list, keys_list_sizes, &values_list, code = tsdbCacheGetValuesFromRocks(pTsdb, num_keys, (const char *const *)keys_list, keys_list_sizes, &values_list,
&values_list_sizes); &values_list_sizes);
@ -1169,7 +1283,11 @@ static int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, SArray
SColVal *pColVal = &updCtx->colVal; SColVal *pColVal = &updCtx->colVal;
SLastCol *pLastCol = NULL; SLastCol *pLastCol = NULL;
(void)tsdbCacheDeserialize(values_list[i], values_list_sizes[i], &pLastCol); code = tsdbCacheDeserialize(values_list[i], values_list_sizes[i], &pLastCol);
if (code != TSDB_CODE_SUCCESS) {
tsdbWarn("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
}
/* /*
if (code) { if (code) {
tsdbError("tsdb/cache: vgId:%d, deserialize failed since %s.", TD_VID(pTsdb->pVnode), tstrerror(code)); tsdbError("tsdb/cache: vgId:%d, deserialize failed since %s.", TD_VID(pTsdb->pVnode), tstrerror(code));
@ -1237,7 +1355,8 @@ _exit:
taosArrayDestroy(remainCols); taosArrayDestroy(remainCols);
if (code) { if (code) {
tsdbError("tsdb/cache: vgId:%d, update failed at line %d since %s.", TD_VID(pTsdb->pVnode), lino, tstrerror(code)); tsdbError("tsdb/cache: vgId:%d, update failed at line %d since %s.", TD_VID(pTsdb->pVnode), __LINE__,
tstrerror(code));
} }
TAOS_RETURN(code); TAOS_RETURN(code);
@ -1268,7 +1387,12 @@ int32_t tsdbCacheRowFormatUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, int6
tsdbRowGetKey(&lRow, &tsdbRowKey); tsdbRowGetKey(&lRow, &tsdbRowKey);
STSDBRowIter iter = {0}; STSDBRowIter iter = {0};
(void)tsdbRowIterOpen(&iter, &lRow, pTSchema); code = tsdbRowIterOpen(&iter, &lRow, pTSchema);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s tsdbRowIterOpen failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
TAOS_CHECK_GOTO(code, &lino, _exit);
}
int32_t iCol = 0; int32_t iCol = 0;
for (SColVal *pColVal = tsdbRowIterNext(&iter); pColVal && iCol < nCol; pColVal = tsdbRowIterNext(&iter), iCol++) { for (SColVal *pColVal = tsdbRowIterNext(&iter); pColVal && iCol < nCol; pColVal = tsdbRowIterNext(&iter), iCol++) {
SLastUpdateCtx updateCtx = {.lflag = LFLAG_LAST_ROW, .tsdbRowKey = tsdbRowKey, .colVal = *pColVal}; SLastUpdateCtx updateCtx = {.lflag = LFLAG_LAST_ROW, .tsdbRowKey = tsdbRowKey, .colVal = *pColVal};
@ -1312,13 +1436,23 @@ int32_t tsdbCacheRowFormatUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, int6
if (!taosArrayPush(ctxArray, &updateCtx)) { if (!taosArrayPush(ctxArray, &updateCtx)) {
TAOS_CHECK_GOTO(terrno, &lino, _exit); TAOS_CHECK_GOTO(terrno, &lino, _exit);
} }
(void)tSimpleHashIterateRemove(iColHash, &iCol, sizeof(iCol), &pIte, &iter); code = tSimpleHashIterateRemove(iColHash, &iCol, sizeof(iCol), &pIte, &iter);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s tSimpleHashIterateRemove failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__,
__LINE__, tstrerror(code));
TAOS_CHECK_GOTO(code, &lino, _exit);
}
} }
} }
} }
// 3. do update // 3. do update
(void)tsdbCacheUpdate(pTsdb, suid, uid, ctxArray); code = tsdbCacheUpdate(pTsdb, suid, uid, ctxArray);
if (code < TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s tsdbCacheUpdate failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
TAOS_CHECK_GOTO(code, &lino, _exit);
}
_exit: _exit:
taosMemoryFreeClear(pTSchema); taosMemoryFreeClear(pTSchema);
@ -1384,7 +1518,12 @@ int32_t tsdbCacheColFormatUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, SBlo
// 2. prepare last row // 2. prepare last row
STSDBRowIter iter = {0}; STSDBRowIter iter = {0};
(void)tsdbRowIterOpen(&iter, &lRow, pTSchema); code = tsdbRowIterOpen(&iter, &lRow, pTSchema);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s tsdbRowIterOpen failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
TAOS_CHECK_GOTO(code, &lino, _exit);
}
for (SColVal *pColVal = tsdbRowIterNext(&iter); pColVal; pColVal = tsdbRowIterNext(&iter)) { for (SColVal *pColVal = tsdbRowIterNext(&iter); pColVal; pColVal = tsdbRowIterNext(&iter)) {
SLastUpdateCtx updateCtx = {.lflag = LFLAG_LAST_ROW, .tsdbRowKey = tsdbRowKey, .colVal = *pColVal}; SLastUpdateCtx updateCtx = {.lflag = LFLAG_LAST_ROW, .tsdbRowKey = tsdbRowKey, .colVal = *pColVal};
if (!taosArrayPush(ctxArray, &updateCtx)) { if (!taosArrayPush(ctxArray, &updateCtx)) {
@ -1394,7 +1533,12 @@ int32_t tsdbCacheColFormatUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, SBlo
tsdbRowClose(&iter); tsdbRowClose(&iter);
// 3. do update // 3. do update
(void)tsdbCacheUpdate(pTsdb, suid, uid, ctxArray); code = tsdbCacheUpdate(pTsdb, suid, uid, ctxArray);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s tsdbCacheUpdate failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
TAOS_CHECK_GOTO(code, &lino, _exit);
}
_exit: _exit:
taosMemoryFreeClear(pTSchema); taosMemoryFreeClear(pTSchema);
@ -1604,7 +1748,7 @@ static int32_t tsdbCacheLoadFromRocks(STsdb *pTsdb, tb_uid_t uid, SArray *pLastA
keys_list_sizes[i] = ROCKS_KEY_LEN; keys_list_sizes[i] = ROCKS_KEY_LEN;
} }
rocksMayWrite(pTsdb, true); // flush writebatch cache rocksMayWrite(pTsdb, true); // flush writebatch cache
code = tsdbCacheGetValuesFromRocks(pTsdb, num_keys, (const char *const *)keys_list, keys_list_sizes, &values_list, code = tsdbCacheGetValuesFromRocks(pTsdb, num_keys, (const char *const *)keys_list, keys_list_sizes, &values_list,
&values_list_sizes); &values_list_sizes);
@ -1624,7 +1768,11 @@ static int32_t tsdbCacheLoadFromRocks(STsdb *pTsdb, tb_uid_t uid, SArray *pLastA
continue; continue;
} }
(void)tsdbCacheDeserialize(values_list[i], values_list_sizes[i], &pLastCol); code = tsdbCacheDeserialize(values_list[i], values_list_sizes[i], &pLastCol);
if (code != TSDB_CODE_SUCCESS) {
tsdbWarn("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
}
SLastCol *pToFree = pLastCol; SLastCol *pToFree = pLastCol;
SIdxKey *idxKey = &((SIdxKey *)TARRAY_DATA(remainCols))[j]; SIdxKey *idxKey = &((SIdxKey *)TARRAY_DATA(remainCols))[j];
if (pLastCol && pLastCol->cacheStatus != TSDB_LAST_CACHE_NO_CACHE) { if (pLastCol && pLastCol->cacheStatus != TSDB_LAST_CACHE_NO_CACHE) {
@ -1757,7 +1905,11 @@ int32_t tsdbCacheGetBatch(STsdb *pTsdb, tb_uid_t uid, SArray *pLastArray, SCache
} }
if (h) { if (h) {
(void)taosLRUCacheRelease(pCache, h, false); code = taosLRUCacheRelease(pCache, h, false);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s release lru cache failed at line %d.", TD_VID(pTsdb->pVnode), __func__, __LINE__);
goto _exit;
}
} }
} }
@ -1786,7 +1938,11 @@ int32_t tsdbCacheGetBatch(STsdb *pTsdb, tb_uid_t uid, SArray *pLastArray, SCache
} }
if (h) { if (h) {
(void)taosLRUCacheRelease(pCache, h, false); code = taosLRUCacheRelease(pCache, h, false);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s release lru cache failed at line %d.", TD_VID(pTsdb->pVnode), __func__, __LINE__);
goto _exit;
}
} }
} }
@ -1820,7 +1976,13 @@ int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKE
int numKeys = 0; int numKeys = 0;
SArray *remainCols = NULL; SArray *remainCols = NULL;
(void)tsdbCacheCommit(pTsdb); code = tsdbCacheCommit(pTsdb);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s commit failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
}
(void)taosThreadMutexLock(&pTsdb->lruMutex); (void)taosThreadMutexLock(&pTsdb->lruMutex);
@ -1837,7 +1999,9 @@ int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKE
.cacheStatus = TSDB_LAST_CACHE_NO_CACHE}; .cacheStatus = TSDB_LAST_CACHE_NO_CACHE};
code = tsdbCachePutToLRU(pTsdb, &lastKey, &noneCol); code = tsdbCachePutToLRU(pTsdb, &lastKey, &noneCol);
} }
(void)taosLRUCacheRelease(pTsdb->lruCache, h, false); if (taosLRUCacheRelease(pTsdb->lruCache, h, false) != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s release lru cache failed at line %d.", TD_VID(pTsdb->pVnode), __func__, __LINE__);
}
TAOS_CHECK_EXIT(code); TAOS_CHECK_EXIT(code);
} else { } else {
if (!remainCols) { if (!remainCols) {
@ -1871,7 +2035,7 @@ int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKE
code = terrno; code = terrno;
goto _exit; goto _exit;
} }
SIdxKey* idxKey = taosArrayGet(remainCols, i); SIdxKey *idxKey = taosArrayGet(remainCols, i);
((SLastKey *)key)[0] = idxKey->key; ((SLastKey *)key)[0] = idxKey->key;
@ -1879,7 +2043,7 @@ int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKE
keys_list_sizes[i] = klen; keys_list_sizes[i] = klen;
} }
rocksMayWrite(pTsdb, true); // flush writebatch cache rocksMayWrite(pTsdb, true); // flush writebatch cache
TAOS_CHECK_GOTO(tsdbCacheGetValuesFromRocks(pTsdb, numKeys, (const char *const *)keys_list, keys_list_sizes, TAOS_CHECK_GOTO(tsdbCacheGetValuesFromRocks(pTsdb, numKeys, (const char *const *)keys_list, keys_list_sizes,
&values_list, &values_list_sizes), &values_list, &values_list_sizes),
@ -1888,8 +2052,12 @@ int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKE
rocksdb_writebatch_t *wb = pTsdb->rCache.writebatch; rocksdb_writebatch_t *wb = pTsdb->rCache.writebatch;
for (int i = 0; i < numKeys; ++i) { for (int i = 0; i < numKeys; ++i) {
SLastCol *pLastCol = NULL; SLastCol *pLastCol = NULL;
(void)tsdbCacheDeserialize(values_list[i], values_list_sizes[i], &pLastCol); code = tsdbCacheDeserialize(values_list[i], values_list_sizes[i], &pLastCol);
SIdxKey* idxKey = taosArrayGet(remainCols, i); if (code != TSDB_CODE_SUCCESS) {
tsdbWarn("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
}
SIdxKey *idxKey = taosArrayGet(remainCols, i);
SLastKey *pLastKey = &idxKey->key; SLastKey *pLastKey = &idxKey->key;
if (NULL != pLastCol && (pLastCol->rowKey.ts <= eKey && pLastCol->rowKey.ts >= sKey)) { if (NULL != pLastCol && (pLastCol->rowKey.ts <= eKey && pLastCol->rowKey.ts >= sKey)) {
SLastCol noCacheCol = {.rowKey.ts = TSKEY_MIN, SLastCol noCacheCol = {.rowKey.ts = TSKEY_MIN,
@ -2389,7 +2557,12 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow, bool *pIgnoreEarlie
state->pr->pCurFileSet = state->pFileSet; state->pr->pCurFileSet = state->pFileSet;
(void)loadDataTomb(state->pr, state->pr->pFileReader); code = loadDataTomb(state->pr, state->pr->pFileReader);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s load tomb failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
TAOS_CHECK_GOTO(code, &lino, _err);
}
TAOS_CHECK_GOTO(tsdbDataFileReadBrinBlk(state->pr->pFileReader, &state->pr->pBlkArray), &lino, _err); TAOS_CHECK_GOTO(tsdbDataFileReadBrinBlk(state->pr->pFileReader, &state->pr->pBlkArray), &lino, _err);
} }
@ -2467,7 +2640,12 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow, bool *pIgnoreEarlie
if (!state->pLastRow) { if (!state->pLastRow) {
if (state->pLastIter) { if (state->pLastIter) {
(void)lastIterClose(&state->pLastIter); code = lastIterClose(&state->pLastIter);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s close last iter failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
TAOS_RETURN(code);
}
} }
clearLastFileSet(state); clearLastFileSet(state);
@ -2575,7 +2753,12 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow, bool *pIgnoreEarlie
if (!state->pLastRow) { if (!state->pLastRow) {
if (state->pLastIter) { if (state->pLastIter) {
(void)lastIterClose(&state->pLastIter); code = lastIterClose(&state->pLastIter);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s close last iter failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
TAOS_RETURN(code);
}
} }
*ppRow = &state->row; *ppRow = &state->row;
@ -2599,7 +2782,12 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow, bool *pIgnoreEarlie
} else { } else {
// TODO: merge rows and *ppRow = mergedRow // TODO: merge rows and *ppRow = mergedRow
SRowMerger *pMerger = &state->rowMerger; SRowMerger *pMerger = &state->rowMerger;
(void)tsdbRowMergerInit(pMerger, state->pTSchema); code = tsdbRowMergerInit(pMerger, state->pTSchema);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s init row merger failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
TAOS_RETURN(code);
}
TAOS_CHECK_GOTO(tsdbRowMergerAdd(pMerger, &state->row, state->pTSchema), &lino, _err); TAOS_CHECK_GOTO(tsdbRowMergerAdd(pMerger, &state->row, state->pTSchema), &lino, _err);
TAOS_CHECK_GOTO(tsdbRowMergerAdd(pMerger, state->pLastRow, state->pTSchema), &lino, _err); TAOS_CHECK_GOTO(tsdbRowMergerAdd(pMerger, state->pLastRow, state->pTSchema), &lino, _err);
@ -2765,7 +2953,11 @@ int32_t clearNextRowFromFS(void *iter) {
} }
if (state->pLastIter) { if (state->pLastIter) {
(void)lastIterClose(&state->pLastIter); code = lastIterClose(&state->pLastIter);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("%s close last iter failed at line %d since %s", __func__, __LINE__, tstrerror(code));
TAOS_RETURN(code);
}
} }
if (state->pBlockData) { if (state->pBlockData) {
@ -2798,7 +2990,11 @@ int32_t clearNextRowFromFS(void *iter) {
static void clearLastFileSet(SFSNextRowIter *state) { static void clearLastFileSet(SFSNextRowIter *state) {
if (state->pLastIter) { if (state->pLastIter) {
(void)lastIterClose(&state->pLastIter); int code = lastIterClose(&state->pLastIter);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("%s close last iter failed at line %d since %s", __func__, __LINE__, tstrerror(code));
return;
}
} }
if (state->pBlockData) { if (state->pBlockData) {
@ -3363,7 +3559,11 @@ _err:
TAOS_RETURN(code); TAOS_RETURN(code);
} }
void tsdbCacheRelease(SLRUCache *pCache, LRUHandle *h) { (void)taosLRUCacheRelease(pCache, h, false); } void tsdbCacheRelease(SLRUCache *pCache, LRUHandle *h) {
if (taosLRUCacheRelease(pCache, h, false)) {
tsdbError("%s release lru cache failed at line %d.", __func__, __LINE__);
}
}
void tsdbCacheSetCapacity(SVnode *pVnode, size_t capacity) { void tsdbCacheSetCapacity(SVnode *pVnode, size_t capacity) {
taosLRUCacheSetCapacity(pVnode->pTsdb->lruCache, capacity); taosLRUCacheSetCapacity(pVnode->pTsdb->lruCache, capacity);

3
source/libs/sync/src/syncAppendEntries.c
View File

@ -175,7 +175,8 @@ _SEND_RESPONSE:
if (accepted && matched) { if (accepted && matched) {
pReply->success = true; pReply->success = true;
// update commit index only after matching // update commit index only after matching
(void)syncNodeUpdateCommitIndex(ths, TMIN(pMsg->commitIndex, pReply->lastSendIndex)); SyncIndex returnIndex = syncNodeUpdateCommitIndex(ths, TMIN(pMsg->commitIndex, pReply->lastSendIndex));
sTrace("vgId:%d, update commit return index %" PRId64 "", ths->vgId, returnIndex);
} }
// ack, i.e. send response // ack, i.e. send response

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

@ -85,10 +85,9 @@ int64_t syncNodeCheckCommitIndex(SSyncNode* ths, SyncIndex indexLikely) {
int32_t code = 0; int32_t code = 0;
if (indexLikely > ths->commitIndex && syncNodeAgreedUpon(ths, indexLikely)) { if (indexLikely > ths->commitIndex && syncNodeAgreedUpon(ths, indexLikely)) {
SyncIndex commitIndex = indexLikely; SyncIndex commitIndex = indexLikely;
// TODO add return when error SyncIndex returnIndex = syncNodeUpdateCommitIndex(ths, commitIndex);
(void)syncNodeUpdateCommitIndex(ths, commitIndex); sTrace("vgId:%d, agreed upon. role:%d, term:%" PRId64 ", index:%" PRId64 ", return:%" PRId64, ths->vgId, ths->state,
sTrace("vgId:%d, agreed upon. role:%d, term:%" PRId64 ", index:%" PRId64 "", ths->vgId, ths->state, raftStoreGetTerm(ths), commitIndex, returnIndex);
raftStoreGetTerm(ths), commitIndex);
} }
return ths->commitIndex; return ths->commitIndex;
} }

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

@ -201,13 +201,13 @@ int32_t syncReconfig(int64_t rid, SSyncCfg* pNewCfg) {
if (pSyncNode->state == TAOS_SYNC_STATE_LEADER || pSyncNode->state == TAOS_SYNC_STATE_ASSIGNED_LEADER) { if (pSyncNode->state == TAOS_SYNC_STATE_LEADER || pSyncNode->state == TAOS_SYNC_STATE_ASSIGNED_LEADER) {
// TODO check return value // TODO check return value
(void)syncNodeStopHeartbeatTimer(pSyncNode); TAOS_CHECK_RETURN(syncNodeStopHeartbeatTimer(pSyncNode));
for (int32_t i = 0; i < TSDB_MAX_REPLICA + TSDB_MAX_LEARNER_REPLICA; ++i) { for (int32_t i = 0; i < TSDB_MAX_REPLICA + TSDB_MAX_LEARNER_REPLICA; ++i) {
(void)syncHbTimerInit(pSyncNode, &pSyncNode->peerHeartbeatTimerArr[i], pSyncNode->replicasId[i]); TAOS_CHECK_RETURN(syncHbTimerInit(pSyncNode, &pSyncNode->peerHeartbeatTimerArr[i], pSyncNode->replicasId[i]));
} }
(void)syncNodeStartHeartbeatTimer(pSyncNode); TAOS_CHECK_RETURN(syncNodeStartHeartbeatTimer(pSyncNode));
// syncNodeReplicate(pSyncNode); // syncNodeReplicate(pSyncNode);
} }
@ -410,9 +410,8 @@ int32_t syncSendTimeoutRsp(int64_t rid, int64_t seq) {
syncNodeRelease(pNode); syncNodeRelease(pNode);
if (ret == 1) { if (ret == 1) {
sInfo("send timeout response, seq:%" PRId64 " handle:%p ahandle:%p", seq, rpcMsg.info.handle, rpcMsg.info.ahandle); sInfo("send timeout response, seq:%" PRId64 " handle:%p ahandle:%p", seq, rpcMsg.info.handle, rpcMsg.info.ahandle);
// TODO check return value code = rpcSendResponse(&rpcMsg);
(void)rpcSendResponse(&rpcMsg); return code;
return 0;
} else { } else {
sError("no message handle to send timeout response, seq:%" PRId64, seq); sError("no message handle to send timeout response, seq:%" PRId64, seq);
return TSDB_CODE_SYN_INTERNAL_ERROR; return TSDB_CODE_SYN_INTERNAL_ERROR;
@ -933,7 +932,7 @@ int32_t syncNodePropose(SSyncNode* pSyncNode, SRpcMsg* pMsg, bool isWeak, int64_
int32_t code = syncBuildClientRequest(&rpcMsg, pMsg, seqNum, isWeak, pSyncNode->vgId); int32_t code = syncBuildClientRequest(&rpcMsg, pMsg, seqNum, isWeak, pSyncNode->vgId);
if (code != 0) { if (code != 0) {
sError("vgId:%d, failed to propose msg while serialize since %s", pSyncNode->vgId, terrstr()); sError("vgId:%d, failed to propose msg while serialize since %s", pSyncNode->vgId, terrstr());
(void)syncRespMgrDel(pSyncNode->pSyncRespMgr, seqNum); code = syncRespMgrDel(pSyncNode->pSyncRespMgr, seqNum);
TAOS_RETURN(code); TAOS_RETURN(code);
} }
@ -941,7 +940,7 @@ int32_t syncNodePropose(SSyncNode* pSyncNode, SRpcMsg* pMsg, bool isWeak, int64_
code = (*pSyncNode->syncEqMsg)(pSyncNode->msgcb, &rpcMsg); code = (*pSyncNode->syncEqMsg)(pSyncNode->msgcb, &rpcMsg);
if (code != 0) { if (code != 0) {
sWarn("vgId:%d, failed to propose msg while enqueue since %s", pSyncNode->vgId, terrstr()); sWarn("vgId:%d, failed to propose msg while enqueue since %s", pSyncNode->vgId, terrstr());
(void)syncRespMgrDel(pSyncNode->pSyncRespMgr, seqNum); TAOS_CHECK_RETURN(syncRespMgrDel(pSyncNode->pSyncRespMgr, seqNum));
} }
if (seq != NULL) *seq = seqNum; if (seq != NULL) *seq = seqNum;
@ -961,7 +960,7 @@ static int32_t syncHbTimerInit(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer, SRa
} }
static int32_t syncHbTimerStart(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer) { static int32_t syncHbTimerStart(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer) {
int32_t ret = 0; int32_t code = 0;
int64_t tsNow = taosGetTimestampMs(); int64_t tsNow = taosGetTimestampMs();
if (syncIsInit()) { if (syncIsInit()) {
SSyncHbTimerData* pData = syncHbTimerDataAcquire(pSyncTimer->hbDataRid); SSyncHbTimerData* pData = syncHbTimerDataAcquire(pSyncTimer->hbDataRid);
@ -980,21 +979,20 @@ static int32_t syncHbTimerStart(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer) {
sTrace("vgId:%d, start hb timer, rid:%" PRId64 " addr:%" PRId64, pSyncNode->vgId, pData->rid, pData->destId.addr); sTrace("vgId:%d, start hb timer, rid:%" PRId64 " addr:%" PRId64, pSyncNode->vgId, pData->rid, pData->destId.addr);
(void)taosTmrReset(pSyncTimer->timerCb, pSyncTimer->timerMS / HEARTBEAT_TICK_NUM, (void*)(pData->rid), TAOS_CHECK_RETURN(taosTmrReset(pSyncTimer->timerCb, pSyncTimer->timerMS / HEARTBEAT_TICK_NUM, (void*)(pData->rid),
syncEnv()->pTimerManager, &pSyncTimer->pTimer); syncEnv()->pTimerManager, &pSyncTimer->pTimer));
} else { } else {
ret = TSDB_CODE_SYN_INTERNAL_ERROR; code = TSDB_CODE_SYN_INTERNAL_ERROR;
sError("vgId:%d, start ctrl hb timer error, sync env is stop", pSyncNode->vgId); sError("vgId:%d, start ctrl hb timer error, sync env is stop", pSyncNode->vgId);
} }
return ret; return code;
} }
static int32_t syncHbTimerStop(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer) { static int32_t syncHbTimerStop(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer) {
int32_t ret = 0; int32_t ret = 0;
(void)atomic_add_fetch_64(&pSyncTimer->logicClock, 1); (void)atomic_add_fetch_64(&pSyncTimer->logicClock, 1);
if (!taosTmrStop(pSyncTimer->pTimer)) { bool stop = taosTmrStop(pSyncTimer->pTimer);
return TSDB_CODE_SYN_INTERNAL_ERROR; sDebug("vgId:%d, stop hb timer stop:%d", pSyncNode->vgId, stop);
}
pSyncTimer->pTimer = NULL; pSyncTimer->pTimer = NULL;
syncHbTimerDataRemove(pSyncTimer->hbDataRid); syncHbTimerDataRemove(pSyncTimer->hbDataRid);
pSyncTimer->hbDataRid = -1; pSyncTimer->hbDataRid = -1;
@ -1141,8 +1139,8 @@ SSyncNode* syncNodeOpen(SSyncInfo* pSyncInfo, int32_t vnodeVersion) {
pSyncNode->replicaNum = pSyncNode->raftCfg.cfg.replicaNum; pSyncNode->replicaNum = pSyncNode->raftCfg.cfg.replicaNum;
pSyncNode->totalReplicaNum = pSyncNode->raftCfg.cfg.totalReplicaNum; pSyncNode->totalReplicaNum = pSyncNode->raftCfg.cfg.totalReplicaNum;
for (int32_t i = 0; i < pSyncNode->raftCfg.cfg.totalReplicaNum; ++i) { for (int32_t i = 0; i < pSyncNode->raftCfg.cfg.totalReplicaNum; ++i) {
if (!syncUtilNodeInfo2RaftId(&pSyncNode->raftCfg.cfg.nodeInfo[i], pSyncNode->vgId, &pSyncNode->replicasId[i])) { if (syncUtilNodeInfo2RaftId(&pSyncNode->raftCfg.cfg.nodeInfo[i], pSyncNode->vgId, &pSyncNode->replicasId[i]) ==
terrno = TSDB_CODE_SYN_INTERNAL_ERROR; false) {
sError("vgId:%d, failed to determine raft member id, replica:%d", pSyncNode->vgId, i); sError("vgId:%d, failed to determine raft member id, replica:%d", pSyncNode->vgId, i);
goto _error; goto _error;
} }
@ -1308,7 +1306,10 @@ SSyncNode* syncNodeOpen(SSyncInfo* pSyncInfo, int32_t vnodeVersion) {
} }
// tools // tools
(void)syncRespMgrCreate(pSyncNode, SYNC_RESP_TTL_MS, &pSyncNode->pSyncRespMgr); // TODO: check return value if ((code = syncRespMgrCreate(pSyncNode, SYNC_RESP_TTL_MS, &pSyncNode->pSyncRespMgr)) != 0) {
sError("vgId:%d, failed to create SyncRespMgr", pSyncNode->vgId);
goto _error;
}
if (pSyncNode->pSyncRespMgr == NULL) { if (pSyncNode->pSyncRespMgr == NULL) {
sError("vgId:%d, failed to create SyncRespMgr", pSyncNode->vgId); sError("vgId:%d, failed to create SyncRespMgr", pSyncNode->vgId);
goto _error; goto _error;
@ -1471,29 +1472,31 @@ int32_t syncNodeStart(SSyncNode* pSyncNode) {
#ifdef BUILD_NO_CALL #ifdef BUILD_NO_CALL
int32_t syncNodeStartStandBy(SSyncNode* pSyncNode) { int32_t syncNodeStartStandBy(SSyncNode* pSyncNode) {
// state change // state change
int32_t code = 0;
pSyncNode->state = TAOS_SYNC_STATE_FOLLOWER; pSyncNode->state = TAOS_SYNC_STATE_FOLLOWER;
pSyncNode->roleTimeMs = taosGetTimestampMs(); pSyncNode->roleTimeMs = taosGetTimestampMs();
// TODO check return value // TODO check return value
(void)syncNodeStopHeartbeatTimer(pSyncNode); TAOS_CHECK_RETURN(syncNodeStopHeartbeatTimer(pSyncNode));
// reset elect timer, long enough // reset elect timer, long enough
int32_t electMS = TIMER_MAX_MS; int32_t electMS = TIMER_MAX_MS;
int32_t ret = syncNodeRestartElectTimer(pSyncNode, electMS); code = syncNodeRestartElectTimer(pSyncNode, electMS);
if (ret < 0) { if (code < 0) {
sError("vgId:%d, failed to restart elect timer since %s", pSyncNode->vgId, terrstr()); sError("vgId:%d, failed to restart elect timer since %s", pSyncNode->vgId, terrstr());
return -1; return -1;
} }
ret = syncNodeStartPingTimer(pSyncNode); code = syncNodeStartPingTimer(pSyncNode);
if (ret < 0) { if (code < 0) {
sError("vgId:%d, failed to start ping timer since %s", pSyncNode->vgId, terrstr()); sError("vgId:%d, failed to start ping timer since %s", pSyncNode->vgId, terrstr());
return -1; return -1;
} }
return ret; return code;
} }
#endif #endif
void syncNodePreClose(SSyncNode* pSyncNode) { void syncNodePreClose(SSyncNode* pSyncNode) {
int32_t code = 0;
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
sError("failed to pre close sync node since sync node is null"); sError("failed to pre close sync node since sync node is null");
return; return;
@ -1508,13 +1511,22 @@ void syncNodePreClose(SSyncNode* pSyncNode) {
} }
// stop elect timer // stop elect timer
(void)syncNodeStopElectTimer(pSyncNode); if ((code = syncNodeStopElectTimer(pSyncNode)) != 0) {
sError("vgId:%d, failed to stop elect timer since %s", pSyncNode->vgId, tstrerror(code));
return;
}
// stop heartbeat timer // stop heartbeat timer
(void)syncNodeStopHeartbeatTimer(pSyncNode); if ((code = syncNodeStopHeartbeatTimer(pSyncNode)) != 0) {
sError("vgId:%d, failed to stop heartbeat timer since %s", pSyncNode->vgId, tstrerror(code));
return;
}
// stop ping timer // stop ping timer
(void)syncNodeStopPingTimer(pSyncNode); if ((code = syncNodeStopPingTimer(pSyncNode)) != 0) {
sError("vgId:%d, failed to stop ping timer since %s", pSyncNode->vgId, tstrerror(code));
return;
}
// clean rsp // clean rsp
syncRespCleanRsp(pSyncNode->pSyncRespMgr); syncRespCleanRsp(pSyncNode->pSyncRespMgr);
@ -1536,14 +1548,24 @@ void syncNodePostClose(SSyncNode* pSyncNode) {
void syncHbTimerDataFree(SSyncHbTimerData* pData) { taosMemoryFree(pData); } void syncHbTimerDataFree(SSyncHbTimerData* pData) { taosMemoryFree(pData); }
void syncNodeClose(SSyncNode* pSyncNode) { void syncNodeClose(SSyncNode* pSyncNode) {
int32_t code = 0;
if (pSyncNode == NULL) return; if (pSyncNode == NULL) return;
sNInfo(pSyncNode, "sync close, node:%p", pSyncNode); sNInfo(pSyncNode, "sync close, node:%p", pSyncNode);
syncRespCleanRsp(pSyncNode->pSyncRespMgr); syncRespCleanRsp(pSyncNode->pSyncRespMgr);
(void)syncNodeStopPingTimer(pSyncNode); if ((code = syncNodeStopPingTimer(pSyncNode)) != 0) {
(void)syncNodeStopElectTimer(pSyncNode); sError("vgId:%d, failed to stop ping timer since %s", pSyncNode->vgId, tstrerror(code));
(void)syncNodeStopHeartbeatTimer(pSyncNode); return;
}
if ((code = syncNodeStopElectTimer(pSyncNode)) != 0) {
sError("vgId:%d, failed to stop elect timer since %s", pSyncNode->vgId, tstrerror(code));
return;
}
if ((code = syncNodeStopHeartbeatTimer(pSyncNode)) != 0) {
sError("vgId:%d, failed to stop heartbeat timer since %s", pSyncNode->vgId, tstrerror(code));
return;
}
syncNodeLogReplDestroy(pSyncNode); syncNodeLogReplDestroy(pSyncNode);
syncRespMgrDestroy(pSyncNode->pSyncRespMgr); syncRespMgrDestroy(pSyncNode->pSyncRespMgr);
@ -1599,28 +1621,28 @@ ESyncStrategy syncNodeStrategy(SSyncNode* pSyncNode) { return pSyncNode->raftCfg
// timer control -------------- // timer control --------------
int32_t syncNodeStartPingTimer(SSyncNode* pSyncNode) { int32_t syncNodeStartPingTimer(SSyncNode* pSyncNode) {
int32_t ret = 0; int32_t code = 0;
if (syncIsInit()) { if (syncIsInit()) {
(void)taosTmrReset(pSyncNode->FpPingTimerCB, pSyncNode->pingTimerMS, (void*)pSyncNode->rid, TAOS_CHECK_RETURN(taosTmrReset(pSyncNode->FpPingTimerCB, pSyncNode->pingTimerMS, (void*)pSyncNode->rid,
syncEnv()->pTimerManager, &pSyncNode->pPingTimer); syncEnv()->pTimerManager, &pSyncNode->pPingTimer));
atomic_store_64(&pSyncNode->pingTimerLogicClock, pSyncNode->pingTimerLogicClockUser); atomic_store_64(&pSyncNode->pingTimerLogicClock, pSyncNode->pingTimerLogicClockUser);
} else { } else {
sError("vgId:%d, start ping timer error, sync env is stop", pSyncNode->vgId); sError("vgId:%d, start ping timer error, sync env is stop", pSyncNode->vgId);
} }
return ret; return code;
} }
int32_t syncNodeStopPingTimer(SSyncNode* pSyncNode) { int32_t syncNodeStopPingTimer(SSyncNode* pSyncNode) {
int32_t ret = 0; int32_t code = 0;
(void)atomic_add_fetch_64(&pSyncNode->pingTimerLogicClockUser, 1); (void)atomic_add_fetch_64(&pSyncNode->pingTimerLogicClockUser, 1);
// TODO check return value bool stop = taosTmrStop(pSyncNode->pPingTimer);
(void)taosTmrStop(pSyncNode->pPingTimer); sDebug("vgId:%d, stop ping timer, stop:%d", pSyncNode->vgId, stop);
pSyncNode->pPingTimer = NULL; pSyncNode->pPingTimer = NULL;
return ret; return code;
} }
int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode, int32_t ms) { int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode, int32_t ms) {
int32_t ret = 0; int32_t code = 0;
if (syncIsInit()) { if (syncIsInit()) {
pSyncNode->electTimerMS = ms; pSyncNode->electTimerMS = ms;
@ -1630,22 +1652,22 @@ int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode, int32_t ms) {
pSyncNode->electTimerParam.pSyncNode = pSyncNode; pSyncNode->electTimerParam.pSyncNode = pSyncNode;
pSyncNode->electTimerParam.pData = NULL; pSyncNode->electTimerParam.pData = NULL;
(void)taosTmrReset(pSyncNode->FpElectTimerCB, pSyncNode->electTimerMS, (void*)(pSyncNode->rid), TAOS_CHECK_RETURN(taosTmrReset(pSyncNode->FpElectTimerCB, pSyncNode->electTimerMS, (void*)(pSyncNode->rid),
syncEnv()->pTimerManager, &pSyncNode->pElectTimer); syncEnv()->pTimerManager, &pSyncNode->pElectTimer));
} else { } else {
sError("vgId:%d, start elect timer error, sync env is stop", pSyncNode->vgId); sError("vgId:%d, start elect timer error, sync env is stop", pSyncNode->vgId);
} }
return ret; return code;
} }
int32_t syncNodeStopElectTimer(SSyncNode* pSyncNode) { int32_t syncNodeStopElectTimer(SSyncNode* pSyncNode) {
int32_t ret = 0; int32_t code = 0;
(void)atomic_add_fetch_64(&pSyncNode->electTimerLogicClock, 1); (void)atomic_add_fetch_64(&pSyncNode->electTimerLogicClock, 1);
// TODO check return value bool stop = taosTmrStop(pSyncNode->pElectTimer);
(void)taosTmrStop(pSyncNode->pElectTimer); sDebug("vgId:%d, stop elect timer, stop:%d", pSyncNode->vgId, stop);
pSyncNode->pElectTimer = NULL; pSyncNode->pElectTimer = NULL;
return ret; return code;
} }
int32_t syncNodeRestartElectTimer(SSyncNode* pSyncNode, int32_t ms) { int32_t syncNodeRestartElectTimer(SSyncNode* pSyncNode, int32_t ms) {
@ -1666,7 +1688,10 @@ void syncNodeResetElectTimer(SSyncNode* pSyncNode) {
} }
// TODO check return value // TODO check return value
(void)syncNodeRestartElectTimer(pSyncNode, electMS); if ((code = syncNodeRestartElectTimer(pSyncNode, electMS)) != 0) {
sError("vgId:%d, failed to restart elect timer since %s", pSyncNode->vgId, terrstr());
return;
};
sNTrace(pSyncNode, "reset elect timer, min:%d, max:%d, ms:%d", pSyncNode->electBaseLine, 2 * pSyncNode->electBaseLine, sNTrace(pSyncNode, "reset elect timer, min:%d, max:%d, ms:%d", pSyncNode->electBaseLine, 2 * pSyncNode->electBaseLine,
electMS); electMS);
@ -1674,17 +1699,17 @@ void syncNodeResetElectTimer(SSyncNode* pSyncNode) {
#ifdef BUILD_NO_CALL #ifdef BUILD_NO_CALL
static int32_t syncNodeDoStartHeartbeatTimer(SSyncNode* pSyncNode) { static int32_t syncNodeDoStartHeartbeatTimer(SSyncNode* pSyncNode) {
int32_t ret = 0; int32_t code = 0;
if (syncIsInit()) { if (syncIsInit()) {
(void)taosTmrReset(pSyncNode->FpHeartbeatTimerCB, pSyncNode->heartbeatTimerMS, (void*)pSyncNode->rid, TAOS_CHECK_RETURN(taosTmrReset(pSyncNode->FpHeartbeatTimerCB, pSyncNode->heartbeatTimerMS, (void*)pSyncNode->rid,
syncEnv()->pTimerManager, &pSyncNode->pHeartbeatTimer); syncEnv()->pTimerManager, &pSyncNode->pHeartbeatTimer));
atomic_store_64(&pSyncNode->heartbeatTimerLogicClock, pSyncNode->heartbeatTimerLogicClockUser); atomic_store_64(&pSyncNode->heartbeatTimerLogicClock, pSyncNode->heartbeatTimerLogicClockUser);
} else { } else {
sError("vgId:%d, start heartbeat timer error, sync env is stop", pSyncNode->vgId); sError("vgId:%d, start heartbeat timer error, sync env is stop", pSyncNode->vgId);
} }
sNTrace(pSyncNode, "start heartbeat timer, ms:%d", pSyncNode->heartbeatTimerMS); sNTrace(pSyncNode, "start heartbeat timer, ms:%d", pSyncNode->heartbeatTimerMS);
return ret; return code;
} }
#endif #endif
@ -1707,12 +1732,12 @@ int32_t syncNodeStartHeartbeatTimer(SSyncNode* pSyncNode) {
} }
int32_t syncNodeStopHeartbeatTimer(SSyncNode* pSyncNode) { int32_t syncNodeStopHeartbeatTimer(SSyncNode* pSyncNode) {
int32_t ret = 0; int32_t code = 0;
#if 0 #if 0
//TODO check return value TAOS_CHECK_RETURN(atomic_add_fetch_64(&pSyncNode->heartbeatTimerLogicClockUser, 1));
(void)atomic_add_fetch_64(&pSyncNode->heartbeatTimerLogicClockUser, 1); bool stop = taosTmrStop(pSyncNode->pHeartbeatTimer);
(void)taosTmrStop(pSyncNode->pHeartbeatTimer); sDebug("vgId:%d, stop heartbeat timer, stop:%d", pSyncNode->vgId, stop);
pSyncNode->pHeartbeatTimer = NULL; pSyncNode->pHeartbeatTimer = NULL;
#endif #endif
@ -1723,14 +1748,15 @@ int32_t syncNodeStopHeartbeatTimer(SSyncNode* pSyncNode) {
} }
} }
return ret; return code;
} }
#ifdef BUILD_NO_CALL #ifdef BUILD_NO_CALL
int32_t syncNodeRestartHeartbeatTimer(SSyncNode* pSyncNode) { int32_t syncNodeRestartHeartbeatTimer(SSyncNode* pSyncNode) {
// TODO check return value // TODO check return value
(void)syncNodeStopHeartbeatTimer(pSyncNode); int32_t code = 0;
(void)syncNodeStartHeartbeatTimer(pSyncNode); TAOS_CHECK_RETURN(syncNodeStopHeartbeatTimer(pSyncNode));
TAOS_CHECK_RETURN(syncNodeStartHeartbeatTimer(pSyncNode));
return 0; return 0;
} }
#endif #endif
@ -1806,6 +1832,7 @@ static bool syncIsConfigChanged(const SSyncCfg* pOldCfg, const SSyncCfg* pNewCfg
} }
int32_t syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* pNewConfig, SyncIndex lastConfigChangeIndex) { int32_t syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* pNewConfig, SyncIndex lastConfigChangeIndex) {
int32_t code = 0;
SSyncCfg oldConfig = pSyncNode->raftCfg.cfg; SSyncCfg oldConfig = pSyncNode->raftCfg.cfg;
if (!syncIsConfigChanged(&oldConfig, pNewConfig)) { if (!syncIsConfigChanged(&oldConfig, pNewConfig)) {
sInfo("vgId:1, sync not reconfig since not changed"); sInfo("vgId:1, sync not reconfig since not changed");
@ -1873,7 +1900,7 @@ int32_t syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* pNewConfig, SyncI
// init internal // init internal
pSyncNode->myNodeInfo = pSyncNode->raftCfg.cfg.nodeInfo[pSyncNode->raftCfg.cfg.myIndex]; pSyncNode->myNodeInfo = pSyncNode->raftCfg.cfg.nodeInfo[pSyncNode->raftCfg.cfg.myIndex];
(void)syncUtilNodeInfo2RaftId(&pSyncNode->myNodeInfo, pSyncNode->vgId, &pSyncNode->myRaftId); if (syncUtilNodeInfo2RaftId(&pSyncNode->myNodeInfo, pSyncNode->vgId, &pSyncNode->myRaftId) == false) return terrno;
// init peersNum, peers, peersId // init peersNum, peers, peersId
pSyncNode->peersNum = pSyncNode->raftCfg.cfg.totalReplicaNum - 1; pSyncNode->peersNum = pSyncNode->raftCfg.cfg.totalReplicaNum - 1;
@ -1886,14 +1913,17 @@ int32_t syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* pNewConfig, SyncI
} }
} }
for (int32_t i = 0; i < pSyncNode->peersNum; ++i) { for (int32_t i = 0; i < pSyncNode->peersNum; ++i) {
(void)syncUtilNodeInfo2RaftId(&pSyncNode->peersNodeInfo[i], pSyncNode->vgId, &pSyncNode->peersId[i]); if (syncUtilNodeInfo2RaftId(&pSyncNode->peersNodeInfo[i], pSyncNode->vgId, &pSyncNode->peersId[i]) == false)
return terrno;
} }
// init replicaNum, replicasId // init replicaNum, replicasId
pSyncNode->replicaNum = pSyncNode->raftCfg.cfg.replicaNum; pSyncNode->replicaNum = pSyncNode->raftCfg.cfg.replicaNum;
pSyncNode->totalReplicaNum = pSyncNode->raftCfg.cfg.totalReplicaNum; pSyncNode->totalReplicaNum = pSyncNode->raftCfg.cfg.totalReplicaNum;
for (int32_t i = 0; i < pSyncNode->raftCfg.cfg.totalReplicaNum; ++i) { for (int32_t i = 0; i < pSyncNode->raftCfg.cfg.totalReplicaNum; ++i) {
(void)syncUtilNodeInfo2RaftId(&pSyncNode->raftCfg.cfg.nodeInfo[i], pSyncNode->vgId, &pSyncNode->replicasId[i]); if (syncUtilNodeInfo2RaftId(&pSyncNode->raftCfg.cfg.nodeInfo[i], pSyncNode->vgId, &pSyncNode->replicasId[i]) ==
false)
return terrno;
} }
// update quorum first // update quorum first
@ -1939,7 +1969,7 @@ int32_t syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* pNewConfig, SyncI
// create new // create new
for (int32_t i = 0; i < TSDB_MAX_REPLICA + TSDB_MAX_LEARNER_REPLICA; ++i) { for (int32_t i = 0; i < TSDB_MAX_REPLICA + TSDB_MAX_LEARNER_REPLICA; ++i) {
if (pSyncNode->senders[i] == NULL) { if (pSyncNode->senders[i] == NULL) {
(void)snapshotSenderCreate(pSyncNode, i, &pSyncNode->senders[i]); TAOS_CHECK_RETURN(snapshotSenderCreate(pSyncNode, i, &pSyncNode->senders[i]));
if (pSyncNode->senders[i] == NULL) { if (pSyncNode->senders[i] == NULL) {
// will be created later while send snapshot // will be created later while send snapshot
sSError(pSyncNode->senders[i], "snapshot sender create failed while reconfig"); sSError(pSyncNode->senders[i], "snapshot sender create failed while reconfig");
@ -1961,10 +1991,10 @@ int32_t syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* pNewConfig, SyncI
} }
// persist cfg // persist cfg
(void)syncWriteCfgFile(pSyncNode); TAOS_CHECK_RETURN(syncWriteCfgFile(pSyncNode));
} else { } else {
// persist cfg // persist cfg
(void)syncWriteCfgFile(pSyncNode); TAOS_CHECK_RETURN(syncWriteCfgFile(pSyncNode));
sNInfo(pSyncNode, "do not config change from %d to %d", oldConfig.totalReplicaNum, pNewConfig->totalReplicaNum); sNInfo(pSyncNode, "do not config change from %d to %d", oldConfig.totalReplicaNum, pNewConfig->totalReplicaNum);
} }
@ -2015,7 +2045,7 @@ void syncNodeStepDown(SSyncNode* pSyncNode, SyncTerm newTerm) {
void syncNodeLeaderChangeRsp(SSyncNode* pSyncNode) { syncRespCleanRsp(pSyncNode->pSyncRespMgr); } void syncNodeLeaderChangeRsp(SSyncNode* pSyncNode) { syncRespCleanRsp(pSyncNode->pSyncRespMgr); }
void syncNodeBecomeFollower(SSyncNode* pSyncNode, const char* debugStr) { void syncNodeBecomeFollower(SSyncNode* pSyncNode, const char* debugStr) {
// maybe clear leader cache int32_t code = 0; // maybe clear leader cache
if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) { if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
pSyncNode->leaderCache = EMPTY_RAFT_ID; pSyncNode->leaderCache = EMPTY_RAFT_ID;
} }
@ -2025,7 +2055,10 @@ void syncNodeBecomeFollower(SSyncNode* pSyncNode, const char* debugStr) {
// state change // state change
pSyncNode->state = TAOS_SYNC_STATE_FOLLOWER; pSyncNode->state = TAOS_SYNC_STATE_FOLLOWER;
pSyncNode->roleTimeMs = taosGetTimestampMs(); pSyncNode->roleTimeMs = taosGetTimestampMs();
(void)syncNodeStopHeartbeatTimer(pSyncNode); if ((code = syncNodeStopHeartbeatTimer(pSyncNode)) != 0) {
sError("vgId:%d, failed to stop heartbeat timer since %s", pSyncNode->vgId, tstrerror(code));
return;
}
// trace log // trace log
sNTrace(pSyncNode, "become follower %s", debugStr); sNTrace(pSyncNode, "become follower %s", debugStr);
@ -2042,7 +2075,10 @@ void syncNodeBecomeFollower(SSyncNode* pSyncNode, const char* debugStr) {
pSyncNode->minMatchIndex = SYNC_INDEX_INVALID; pSyncNode->minMatchIndex = SYNC_INDEX_INVALID;
// reset log buffer // reset log buffer
(void)syncLogBufferReset(pSyncNode->pLogBuf, pSyncNode); if ((code = syncLogBufferReset(pSyncNode->pLogBuf, pSyncNode)) != 0) {
sError("vgId:%d, failed to reset log buffer since %s", pSyncNode->vgId, tstrerror(code));
return;
}
// reset elect timer // reset elect timer
syncNodeResetElectTimer(pSyncNode); syncNodeResetElectTimer(pSyncNode);
@ -2069,7 +2105,11 @@ void syncNodeBecomeLearner(SSyncNode* pSyncNode, const char* debugStr) {
pSyncNode->minMatchIndex = SYNC_INDEX_INVALID; pSyncNode->minMatchIndex = SYNC_INDEX_INVALID;
// reset log buffer // reset log buffer
(void)syncLogBufferReset(pSyncNode->pLogBuf, pSyncNode); int32_t code = 0;
if ((code = syncLogBufferReset(pSyncNode->pLogBuf, pSyncNode)) != 0) {
sError("vgId:%d, failed to reset log buffer since %s", pSyncNode->vgId, tstrerror(code));
return;
};
} }
// TLA+ Spec // TLA+ Spec
@ -2091,6 +2131,7 @@ void syncNodeBecomeLearner(SSyncNode* pSyncNode, const char* debugStr) {
// /\ UNCHANGED <<messages, currentTerm, votedFor, candidateVars, logVars>> // /\ UNCHANGED <<messages, currentTerm, votedFor, candidateVars, logVars>>
// //
void syncNodeBecomeLeader(SSyncNode* pSyncNode, const char* debugStr) { void syncNodeBecomeLeader(SSyncNode* pSyncNode, const char* debugStr) {
int32_t code = 0;
pSyncNode->becomeLeaderNum++; pSyncNode->becomeLeaderNum++;
pSyncNode->hbrSlowNum = 0; pSyncNode->hbrSlowNum = 0;
@ -2122,7 +2163,10 @@ void syncNodeBecomeLeader(SSyncNode* pSyncNode, const char* debugStr) {
} }
// init peer mgr // init peer mgr
(void)syncNodePeerStateInit(pSyncNode); if ((code = syncNodePeerStateInit(pSyncNode)) != 0) {
sError("vgId:%d, failed to init peer state since %s", pSyncNode->vgId, tstrerror(code));
return;
}
#if 0 #if 0
// update sender private term // update sender private term
@ -2143,13 +2187,22 @@ void syncNodeBecomeLeader(SSyncNode* pSyncNode, const char* debugStr) {
} }
// stop elect timer // stop elect timer
(void)syncNodeStopElectTimer(pSyncNode); if ((code = syncNodeStopElectTimer(pSyncNode)) != 0) {
sError("vgId:%d, failed to stop elect timer since %s", pSyncNode->vgId, tstrerror(code));
return;
}
// start heartbeat timer // start heartbeat timer
(void)syncNodeStartHeartbeatTimer(pSyncNode); if ((code = syncNodeStartHeartbeatTimer(pSyncNode)) != 0) {
sError("vgId:%d, failed to start heartbeat timer since %s", pSyncNode->vgId, tstrerror(code));
return;
}
// send heartbeat right now // send heartbeat right now
(void)syncNodeHeartbeatPeers(pSyncNode); if ((code = syncNodeHeartbeatPeers(pSyncNode)) != 0) {
sError("vgId:%d, failed to send heartbeat to peers since %s", pSyncNode->vgId, tstrerror(code));
return;
}
// call back // call back
if (pSyncNode->pFsm != NULL && pSyncNode->pFsm->FpBecomeLeaderCb != NULL) { if (pSyncNode->pFsm != NULL && pSyncNode->pFsm->FpBecomeLeaderCb != NULL) {
@ -2160,13 +2213,17 @@ void syncNodeBecomeLeader(SSyncNode* pSyncNode, const char* debugStr) {
pSyncNode->minMatchIndex = SYNC_INDEX_INVALID; pSyncNode->minMatchIndex = SYNC_INDEX_INVALID;
// reset log buffer // reset log buffer
(void)syncLogBufferReset(pSyncNode->pLogBuf, pSyncNode); if ((code = syncLogBufferReset(pSyncNode->pLogBuf, pSyncNode)) != 0) {
sError("vgId:%d, failed to reset log buffer since %s", pSyncNode->vgId, tstrerror(code));
return;
}
// trace log // trace log
sNInfo(pSyncNode, "become leader %s", debugStr); sNInfo(pSyncNode, "become leader %s", debugStr);
} }
void syncNodeBecomeAssignedLeader(SSyncNode* pSyncNode) { void syncNodeBecomeAssignedLeader(SSyncNode* pSyncNode) {
int32_t code = 0;
pSyncNode->becomeAssignedLeaderNum++; pSyncNode->becomeAssignedLeaderNum++;
pSyncNode->hbrSlowNum = 0; pSyncNode->hbrSlowNum = 0;
@ -2198,7 +2255,10 @@ void syncNodeBecomeAssignedLeader(SSyncNode* pSyncNode) {
} }
// init peer mgr // init peer mgr
(void)syncNodePeerStateInit(pSyncNode); if ((code = syncNodePeerStateInit(pSyncNode)) != 0) {
sError("vgId:%d, failed to init peer state since %s", pSyncNode->vgId, tstrerror(code));
return;
}
// close receiver // close receiver
if (snapshotReceiverIsStart(pSyncNode->pNewNodeReceiver)) { if (snapshotReceiverIsStart(pSyncNode->pNewNodeReceiver)) {
@ -2206,13 +2266,22 @@ void syncNodeBecomeAssignedLeader(SSyncNode* pSyncNode) {
} }
// stop elect timer // stop elect timer
(void)syncNodeStopElectTimer(pSyncNode); if ((code = syncNodeStopElectTimer(pSyncNode)) != 0) {
sError("vgId:%d, failed to stop elect timer since %s", pSyncNode->vgId, tstrerror(code));
return;
}
// start heartbeat timer // start heartbeat timer
(void)syncNodeStartHeartbeatTimer(pSyncNode); if ((code = syncNodeStartHeartbeatTimer(pSyncNode)) != 0) {
sError("vgId:%d, failed to start heartbeat timer since %s", pSyncNode->vgId, tstrerror(code));
return;
}
// send heartbeat right now // send heartbeat right now
(void)syncNodeHeartbeatPeers(pSyncNode); if ((code = syncNodeHeartbeatPeers(pSyncNode)) != 0) {
sError("vgId:%d, failed to send heartbeat to peers since %s", pSyncNode->vgId, tstrerror(code));
return;
}
// call back // call back
if (pSyncNode->pFsm != NULL && pSyncNode->pFsm->FpBecomeAssignedLeaderCb != NULL) { if (pSyncNode->pFsm != NULL && pSyncNode->pFsm->FpBecomeAssignedLeaderCb != NULL) {
@ -2223,7 +2292,10 @@ void syncNodeBecomeAssignedLeader(SSyncNode* pSyncNode) {
pSyncNode->minMatchIndex = SYNC_INDEX_INVALID; pSyncNode->minMatchIndex = SYNC_INDEX_INVALID;
// reset log buffer // reset log buffer
(void)syncLogBufferReset(pSyncNode->pLogBuf, pSyncNode); if ((code = syncLogBufferReset(pSyncNode->pLogBuf, pSyncNode)) != 0) {
sError("vgId:%d, failed to reset log buffer since %s", pSyncNode->vgId, tstrerror(code));
return;
}
// trace log // trace log
sNInfo(pSyncNode, "become assigned leader"); sNInfo(pSyncNode, "become assigned leader");
@ -2513,8 +2585,10 @@ static void syncNodeEqPingTimer(void* param, void* tmrId) {
} }
_out: _out:
(void)taosTmrReset(syncNodeEqPingTimer, pNode->pingTimerMS, (void*)pNode->rid, syncEnv()->pTimerManager, if ((code = taosTmrReset(syncNodeEqPingTimer, pNode->pingTimerMS, (void*)pNode->rid, syncEnv()->pTimerManager,
&pNode->pPingTimer); &pNode->pPingTimer)) != 0) {
sError("failed to reset ping timer since %s", tstrerror(code));
};
} }
syncNodeRelease(pNode); syncNodeRelease(pNode);
} }
@ -2591,8 +2665,9 @@ static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) {
} }
_out: _out:
(void)taosTmrReset(syncNodeEqHeartbeatTimer, pNode->heartbeatTimerMS, (void*)pNode->rid, syncEnv()->pTimerManager, if (taosTmrReset(syncNodeEqHeartbeatTimer, pNode->heartbeatTimerMS, (void*)pNode->rid, syncEnv()->pTimerManager,
&pNode->pHeartbeatTimer); &pNode->pHeartbeatTimer) != 0)
return;
} else { } else {
sTrace("==syncNodeEqHeartbeatTimer== heartbeatTimerLogicClock:%" PRId64 ", heartbeatTimerLogicClockUser:%" PRId64, sTrace("==syncNodeEqHeartbeatTimer== heartbeatTimerLogicClock:%" PRId64 ", heartbeatTimerLogicClockUser:%" PRId64,
@ -2603,6 +2678,7 @@ static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) {
#endif #endif
static void syncNodeEqPeerHeartbeatTimer(void* param, void* tmrId) { static void syncNodeEqPeerHeartbeatTimer(void* param, void* tmrId) {
int32_t code = 0;
int64_t hbDataRid = (int64_t)param; int64_t hbDataRid = (int64_t)param;
int64_t tsNow = taosGetTimestampMs(); int64_t tsNow = taosGetTimestampMs();
@ -2646,7 +2722,12 @@ static void syncNodeEqPeerHeartbeatTimer(void* param, void* tmrId) {
pData->execTime += pSyncTimer->timerMS; pData->execTime += pSyncTimer->timerMS;
SRpcMsg rpcMsg = {0}; SRpcMsg rpcMsg = {0};
(void)syncBuildHeartbeat(&rpcMsg, pSyncNode->vgId); if ((code = syncBuildHeartbeat(&rpcMsg, pSyncNode->vgId)) != 0) {
sError("vgId:%d, failed to build heartbeat msg since %s", pSyncNode->vgId, tstrerror(code));
syncNodeRelease(pSyncNode);
syncHbTimerDataRelease(pData);
return;
}
pSyncNode->minMatchIndex = syncMinMatchIndex(pSyncNode); pSyncNode->minMatchIndex = syncMinMatchIndex(pSyncNode);
@ -2668,14 +2749,22 @@ static void syncNodeEqPeerHeartbeatTimer(void* param, void* tmrId) {
STraceId* trace = &(rpcMsg.info.traceId); STraceId* trace = &(rpcMsg.info.traceId);
sGTrace("vgId:%d, send sync-heartbeat to dnode:%d", pSyncNode->vgId, DID(&(pSyncMsg->destId))); sGTrace("vgId:%d, send sync-heartbeat to dnode:%d", pSyncNode->vgId, DID(&(pSyncMsg->destId)));
syncLogSendHeartbeat(pSyncNode, pSyncMsg, false, timerElapsed, pData->execTime); syncLogSendHeartbeat(pSyncNode, pSyncMsg, false, timerElapsed, pData->execTime);
(void)syncNodeSendHeartbeat(pSyncNode, &pSyncMsg->destId, &rpcMsg); int ret = syncNodeSendHeartbeat(pSyncNode, &pSyncMsg->destId, &rpcMsg);
if (ret != 0) {
sError("vgId:%d, failed to send heartbeat since %s", pSyncNode->vgId, tstrerror(ret));
}
} else { } else {
} }
if (syncIsInit()) { if (syncIsInit()) {
// sTrace("vgId:%d, reset peer hb timer", pSyncNode->vgId); // sTrace("vgId:%d, reset peer hb timer", pSyncNode->vgId);
(void)taosTmrReset(syncNodeEqPeerHeartbeatTimer, pSyncTimer->timerMS / HEARTBEAT_TICK_NUM, (void*)hbDataRid, if ((code = taosTmrReset(syncNodeEqPeerHeartbeatTimer, pSyncTimer->timerMS / HEARTBEAT_TICK_NUM,
syncEnv()->pTimerManager, &pSyncTimer->pTimer); (void*)hbDataRid, syncEnv()->pTimerManager, &pSyncTimer->pTimer)) != 0) {
sError("vgId:%d, reset peer hb timer error, %s", pSyncNode->vgId, tstrerror(code));
syncNodeRelease(pSyncNode);
syncHbTimerDataRelease(pData);
return;
}
} else { } else {
sError("sync env is stop, reset peer hb timer error"); sError("sync env is stop, reset peer hb timer error");
} }
@ -2715,6 +2804,7 @@ int32_t syncCacheEntry(SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry, LRUHand
void syncBuildConfigFromReq(SAlterVnodeReplicaReq* pReq, SSyncCfg* cfg) { // TODO SAlterVnodeReplicaReq name is proper? void syncBuildConfigFromReq(SAlterVnodeReplicaReq* pReq, SSyncCfg* cfg) { // TODO SAlterVnodeReplicaReq name is proper?
cfg->replicaNum = 0; cfg->replicaNum = 0;
cfg->totalReplicaNum = 0; cfg->totalReplicaNum = 0;
int32_t code = 0;
for (int i = 0; i < pReq->replica; ++i) { for (int i = 0; i < pReq->replica; ++i) {
SNodeInfo* pNode = &cfg->nodeInfo[i]; SNodeInfo* pNode = &cfg->nodeInfo[i];
@ -2722,9 +2812,9 @@ void syncBuildConfigFromReq(SAlterVnodeReplicaReq* pReq, SSyncCfg* cfg) { // TO
pNode->nodePort = pReq->replicas[i].port; pNode->nodePort = pReq->replicas[i].port;
tstrncpy(pNode->nodeFqdn, pReq->replicas[i].fqdn, sizeof(pNode->nodeFqdn)); tstrncpy(pNode->nodeFqdn, pReq->replicas[i].fqdn, sizeof(pNode->nodeFqdn));
pNode->nodeRole = TAOS_SYNC_ROLE_VOTER; pNode->nodeRole = TAOS_SYNC_ROLE_VOTER;
(void)tmsgUpdateDnodeInfo(&pNode->nodeId, &pNode->clusterId, pNode->nodeFqdn, &pNode->nodePort); bool update = tmsgUpdateDnodeInfo(&pNode->nodeId, &pNode->clusterId, pNode->nodeFqdn, &pNode->nodePort);
sInfo("vgId:%d, replica:%d ep:%s:%u dnode:%d nodeRole:%d", pReq->vgId, i, pNode->nodeFqdn, pNode->nodePort, sInfo("vgId:%d, replica:%d ep:%s:%u dnode:%d nodeRole:%d, update:%d", pReq->vgId, i, pNode->nodeFqdn,
pNode->nodeId, pNode->nodeRole); pNode->nodePort, pNode->nodeId, pNode->nodeRole, update);
cfg->replicaNum++; cfg->replicaNum++;
} }
if (pReq->selfIndex != -1) { if (pReq->selfIndex != -1) {
@ -2736,9 +2826,9 @@ void syncBuildConfigFromReq(SAlterVnodeReplicaReq* pReq, SSyncCfg* cfg) { // TO
pNode->nodePort = pReq->learnerReplicas[cfg->totalReplicaNum].port; pNode->nodePort = pReq->learnerReplicas[cfg->totalReplicaNum].port;
pNode->nodeRole = TAOS_SYNC_ROLE_LEARNER; pNode->nodeRole = TAOS_SYNC_ROLE_LEARNER;
tstrncpy(pNode->nodeFqdn, pReq->learnerReplicas[cfg->totalReplicaNum].fqdn, sizeof(pNode->nodeFqdn)); tstrncpy(pNode->nodeFqdn, pReq->learnerReplicas[cfg->totalReplicaNum].fqdn, sizeof(pNode->nodeFqdn));
(void)tmsgUpdateDnodeInfo(&pNode->nodeId, &pNode->clusterId, pNode->nodeFqdn, &pNode->nodePort); bool update = tmsgUpdateDnodeInfo(&pNode->nodeId, &pNode->clusterId, pNode->nodeFqdn, &pNode->nodePort);
sInfo("vgId:%d, replica:%d ep:%s:%u dnode:%d nodeRole:%d", pReq->vgId, i, pNode->nodeFqdn, pNode->nodePort, sInfo("vgId:%d, replica:%d ep:%s:%u dnode:%d nodeRole:%d, update:%d", pReq->vgId, i, pNode->nodeFqdn,
pNode->nodeId, pNode->nodeRole); pNode->nodePort, pNode->nodeId, pNode->nodeRole, update);
cfg->totalReplicaNum++; cfg->totalReplicaNum++;
} }
cfg->totalReplicaNum += pReq->replica; cfg->totalReplicaNum += pReq->replica;
@ -2849,9 +2939,9 @@ int32_t syncNodeRebuildPeerAndCfg(SSyncNode* ths, SSyncCfg* cfg) {
syncUtilNodeInfo2EpSet(&ths->peersNodeInfo[i], &ths->peersEpset[i]); syncUtilNodeInfo2EpSet(&ths->peersNodeInfo[i], &ths->peersEpset[i]);
if (!syncUtilNodeInfo2RaftId(&ths->peersNodeInfo[i], ths->vgId, &ths->peersId[i])) { if (syncUtilNodeInfo2RaftId(&ths->peersNodeInfo[i], ths->vgId, &ths->peersId[i]) == false) {
sError("vgId:%d, failed to determine raft member id, peer:%d", ths->vgId, i); sError("vgId:%d, failed to determine raft member id, peer:%d", ths->vgId, i);
return -1; return terrno;
} }
i++; i++;
@ -2919,8 +3009,7 @@ int32_t syncNodeRebuildAndCopyIfExist(SSyncNode* ths, int32_t oldtotalReplicaNum
ths->replicaNum = ths->raftCfg.cfg.replicaNum; ths->replicaNum = ths->raftCfg.cfg.replicaNum;
ths->totalReplicaNum = ths->raftCfg.cfg.totalReplicaNum; ths->totalReplicaNum = ths->raftCfg.cfg.totalReplicaNum;
for (int32_t i = 0; i < ths->raftCfg.cfg.totalReplicaNum; ++i) { for (int32_t i = 0; i < ths->raftCfg.cfg.totalReplicaNum; ++i) {
if (!syncUtilNodeInfo2RaftId(&ths->raftCfg.cfg.nodeInfo[i], ths->vgId, &ths->replicasId[i])) if (syncUtilNodeInfo2RaftId(&ths->raftCfg.cfg.nodeInfo[i], ths->vgId, &ths->replicasId[i]) == false) return terrno;
return TSDB_CODE_SYN_INTERNAL_ERROR;
} }
// 2.rebuild MatchIndex, remove deleted one // 2.rebuild MatchIndex, remove deleted one
@ -3287,7 +3376,10 @@ int32_t syncNodeAppend(SSyncNode* ths, SSyncRaftEntry* pEntry) {
// append to log buffer // append to log buffer
if ((code = syncLogBufferAppend(ths->pLogBuf, ths, pEntry)) < 0) { if ((code = syncLogBufferAppend(ths->pLogBuf, ths, pEntry)) < 0) {
sError("vgId:%d, failed to enqueue sync log buffer, index:%" PRId64, ths->vgId, pEntry->index); sError("vgId:%d, failed to enqueue sync log buffer, index:%" PRId64, ths->vgId, pEntry->index);
(void)syncFsmExecute(ths, ths->pFsm, ths->state, raftStoreGetTerm(ths), pEntry, terrno, false); int32_t ret = 0;
if ((ret = syncFsmExecute(ths, ths->pFsm, ths->state, raftStoreGetTerm(ths), pEntry, terrno, false)) != 0) {
sError("vgId:%d, failed to execute fsm, since %s", ths->vgId, tstrerror(ret));
}
syncEntryDestroy(pEntry); syncEntryDestroy(pEntry);
pEntry = NULL; pEntry = NULL;
goto _out; goto _out;
@ -3305,7 +3397,7 @@ _out:;
ths->pLogBuf->matchIndex, ths->pLogBuf->endIndex); ths->pLogBuf->matchIndex, ths->pLogBuf->endIndex);
if (code == 0 && ths->state == TAOS_SYNC_STATE_ASSIGNED_LEADER) { if (code == 0 && ths->state == TAOS_SYNC_STATE_ASSIGNED_LEADER) {
(void)syncNodeUpdateAssignedCommitIndex(ths, matchIndex); TAOS_CHECK_RETURN(syncNodeUpdateAssignedCommitIndex(ths, matchIndex));
if (ths->fsmState != SYNC_FSM_STATE_INCOMPLETE && if (ths->fsmState != SYNC_FSM_STATE_INCOMPLETE &&
syncLogBufferCommit(ths->pLogBuf, ths, ths->assignedCommitIndex) < 0) { syncLogBufferCommit(ths->pLogBuf, ths, ths->assignedCommitIndex) < 0) {
@ -3320,7 +3412,8 @@ _out:;
} }
// single replica // single replica
(void)syncNodeUpdateCommitIndex(ths, matchIndex); SyncIndex returnIndex = syncNodeUpdateCommitIndex(ths, matchIndex);
sTrace("vgId:%d, update commit return index %" PRId64 "", ths->vgId, returnIndex);
if (ths->fsmState != SYNC_FSM_STATE_INCOMPLETE && if (ths->fsmState != SYNC_FSM_STATE_INCOMPLETE &&
(code = syncLogBufferCommit(ths->pLogBuf, ths, ths->commitIndex)) < 0) { (code = syncLogBufferCommit(ths->pLogBuf, ths, ths->commitIndex)) < 0) {
@ -3442,7 +3535,7 @@ int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
} }
SRpcMsg rpcMsg = {0}; SRpcMsg rpcMsg = {0};
(void)syncBuildHeartbeatReply(&rpcMsg, ths->vgId); TAOS_CHECK_RETURN(syncBuildHeartbeatReply(&rpcMsg, ths->vgId));
SyncTerm currentTerm = raftStoreGetTerm(ths); SyncTerm currentTerm = raftStoreGetTerm(ths);
SyncHeartbeatReply* pMsgReply = rpcMsg.pCont; SyncHeartbeatReply* pMsgReply = rpcMsg.pCont;
@ -3470,7 +3563,7 @@ int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
if (ths->state == TAOS_SYNC_STATE_FOLLOWER || ths->state == TAOS_SYNC_STATE_LEARNER) { if (ths->state == TAOS_SYNC_STATE_FOLLOWER || ths->state == TAOS_SYNC_STATE_LEARNER) {
SRpcMsg rpcMsgLocalCmd = {0}; SRpcMsg rpcMsgLocalCmd = {0};
(void)syncBuildLocalCmd(&rpcMsgLocalCmd, ths->vgId); TAOS_CHECK_RETURN(syncBuildLocalCmd(&rpcMsgLocalCmd, ths->vgId));
SyncLocalCmd* pSyncMsg = rpcMsgLocalCmd.pCont; SyncLocalCmd* pSyncMsg = rpcMsgLocalCmd.pCont;
pSyncMsg->cmd = pSyncMsg->cmd =
@ -3494,7 +3587,7 @@ int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
if (pMsg->term >= currentTerm && if (pMsg->term >= currentTerm &&
(ths->state == TAOS_SYNC_STATE_LEADER || ths->state == TAOS_SYNC_STATE_ASSIGNED_LEADER)) { (ths->state == TAOS_SYNC_STATE_LEADER || ths->state == TAOS_SYNC_STATE_ASSIGNED_LEADER)) {
SRpcMsg rpcMsgLocalCmd = {0}; SRpcMsg rpcMsgLocalCmd = {0};
(void)syncBuildLocalCmd(&rpcMsgLocalCmd, ths->vgId); TAOS_CHECK_RETURN(syncBuildLocalCmd(&rpcMsgLocalCmd, ths->vgId));
SyncLocalCmd* pSyncMsg = rpcMsgLocalCmd.pCont; SyncLocalCmd* pSyncMsg = rpcMsgLocalCmd.pCont;
pSyncMsg->cmd = SYNC_LOCAL_CMD_STEP_DOWN; pSyncMsg->cmd = SYNC_LOCAL_CMD_STEP_DOWN;
@ -3577,7 +3670,8 @@ int32_t syncNodeOnLocalCmd(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
return TSDB_CODE_SYN_INTERNAL_ERROR; return TSDB_CODE_SYN_INTERNAL_ERROR;
} }
if (pMsg->currentTerm == matchTerm) { if (pMsg->currentTerm == matchTerm) {
(void)syncNodeUpdateCommitIndex(ths, pMsg->commitIndex); SyncIndex returnIndex = syncNodeUpdateCommitIndex(ths, pMsg->commitIndex);
sTrace("vgId:%d, update commit return index %" PRId64 "", ths->vgId, returnIndex);
} }
if (ths->fsmState != SYNC_FSM_STATE_INCOMPLETE && syncLogBufferCommit(ths->pLogBuf, ths, ths->commitIndex) < 0) { if (ths->fsmState != SYNC_FSM_STATE_INCOMPLETE && syncLogBufferCommit(ths->pLogBuf, ths, ths->commitIndex) < 0) {
sError("vgId:%d, failed to commit raft log since %s. commit index:%" PRId64 "", ths->vgId, terrstr(), sError("vgId:%d, failed to commit raft log since %s. commit index:%" PRId64 "", ths->vgId, terrstr(),

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

@ -163,7 +163,7 @@ int32_t syncWriteCfgFile(SSyncNode *pNode) {
TAOS_CHECK_EXIT(TAOS_SYSTEM_ERROR(errno)); TAOS_CHECK_EXIT(TAOS_SYSTEM_ERROR(errno));
} }
(void)taosCloseFile(&pFile); TAOS_CHECK_EXIT(taosCloseFile(&pFile));
TAOS_CHECK_EXIT(taosRenameFile(file, realfile)); TAOS_CHECK_EXIT(taosRenameFile(file, realfile));
sInfo("vgId:%d, succeed to write sync cfg file:%s, len:%d, lastConfigIndex:%" PRId64 ", changeVersion:%d", sInfo("vgId:%d, succeed to write sync cfg file:%s, len:%d, lastConfigIndex:%" PRId64 ", changeVersion:%d",

22
source/libs/sync/src/syncRaftStore.c
View File

@ -134,7 +134,7 @@ int32_t raftStoreWriteFile(SSyncNode *pNode) {
if (taosFsyncFile(pFile) < 0) TAOS_CHECK_GOTO(terrno, &lino, _OVER); if (taosFsyncFile(pFile) < 0) TAOS_CHECK_GOTO(terrno, &lino, _OVER);
(void)taosCloseFile(&pFile); TAOS_CHECK_GOTO(taosCloseFile(&pFile), &lino, _OVER);
if (taosRenameFile(file, realfile) != 0) TAOS_CHECK_GOTO(terrno, &lino, _OVER); if (taosRenameFile(file, realfile) != 0) TAOS_CHECK_GOTO(terrno, &lino, _OVER);
code = 0; code = 0;
@ -168,21 +168,30 @@ bool raftStoreHasVoted(SSyncNode *pNode) {
void raftStoreVote(SSyncNode *pNode, SRaftId *pRaftId) { void raftStoreVote(SSyncNode *pNode, SRaftId *pRaftId) {
(void)taosThreadMutexLock(&pNode->raftStore.mutex); (void)taosThreadMutexLock(&pNode->raftStore.mutex);
pNode->raftStore.voteFor = *pRaftId; pNode->raftStore.voteFor = *pRaftId;
(void)raftStoreWriteFile(pNode); int32_t code = 0;
if ((code = raftStoreWriteFile(pNode)) != 0) {
sError("vgId:%d, failed to write raft store file since %s", pNode->vgId, tstrerror(code));
}
(void)taosThreadMutexUnlock(&pNode->raftStore.mutex); (void)taosThreadMutexUnlock(&pNode->raftStore.mutex);
} }
void raftStoreClearVote(SSyncNode *pNode) { void raftStoreClearVote(SSyncNode *pNode) {
(void)taosThreadMutexLock(&pNode->raftStore.mutex); (void)taosThreadMutexLock(&pNode->raftStore.mutex);
pNode->raftStore.voteFor = EMPTY_RAFT_ID; pNode->raftStore.voteFor = EMPTY_RAFT_ID;
(void)raftStoreWriteFile(pNode); int32_t code = 0;
if ((code = raftStoreWriteFile(pNode)) != 0) {
sError("vgId:%d, failed to write raft store file since %s", pNode->vgId, tstrerror(code));
}
(void)taosThreadMutexUnlock(&pNode->raftStore.mutex); (void)taosThreadMutexUnlock(&pNode->raftStore.mutex);
} }
void raftStoreNextTerm(SSyncNode *pNode) { void raftStoreNextTerm(SSyncNode *pNode) {
(void)taosThreadMutexLock(&pNode->raftStore.mutex); (void)taosThreadMutexLock(&pNode->raftStore.mutex);
pNode->raftStore.currentTerm++; pNode->raftStore.currentTerm++;
(void)raftStoreWriteFile(pNode); int32_t code = 0;
if ((code = raftStoreWriteFile(pNode)) != 0) {
sError("vgId:%d, failed to write raft store file since %s", pNode->vgId, tstrerror(code));
}
(void)taosThreadMutexUnlock(&pNode->raftStore.mutex); (void)taosThreadMutexUnlock(&pNode->raftStore.mutex);
} }
@ -190,7 +199,10 @@ void raftStoreSetTerm(SSyncNode *pNode, SyncTerm term) {
(void)taosThreadMutexLock(&pNode->raftStore.mutex); (void)taosThreadMutexLock(&pNode->raftStore.mutex);
if (pNode->raftStore.currentTerm < term) { if (pNode->raftStore.currentTerm < term) {
pNode->raftStore.currentTerm = term; pNode->raftStore.currentTerm = term;
(void)raftStoreWriteFile(pNode); int32_t code = 0;
if ((code = raftStoreWriteFile(pNode)) != 0) {
sError("vgId:%d, failed to write raft store file since %s", pNode->vgId, tstrerror(code));
}
} }
(void)taosThreadMutexUnlock(&pNode->raftStore.mutex); (void)taosThreadMutexUnlock(&pNode->raftStore.mutex);
} }

5
source/libs/sync/src/syncReplication.c
View File

@ -118,8 +118,9 @@ int32_t syncNodeHeartbeatPeers(SSyncNode* pSyncNode) {
STraceId* trace = &(rpcMsg.info.traceId); STraceId* trace = &(rpcMsg.info.traceId);
sGTrace("vgId:%d, send sync-heartbeat to dnode:%d", pSyncNode->vgId, DID(&(pSyncMsg->destId))); sGTrace("vgId:%d, send sync-heartbeat to dnode:%d", pSyncNode->vgId, DID(&(pSyncMsg->destId)));
syncLogSendHeartbeat(pSyncNode, pSyncMsg, true, 0, 0); syncLogSendHeartbeat(pSyncNode, pSyncMsg, true, 0, 0);
if (syncNodeSendHeartbeat(pSyncNode, &pSyncMsg->destId, &rpcMsg) != 0) { int32_t ret = syncNodeSendHeartbeat(pSyncNode, &pSyncMsg->destId, &rpcMsg);
sError("vgId:%d, failed to send sync-heartbeat to dnode:%d", pSyncNode->vgId, DID(&(pSyncMsg->destId))); if (ret != 0) {
sError("vgId:%d, failed to send sync-heartbeat since %s", pSyncNode->vgId, tstrerror(ret));
} }
} }

2
source/libs/sync/src/syncRequestVote.c
View File

@ -137,7 +137,7 @@ int32_t syncNodeOnRequestVote(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
// trace log // trace log
syncLogRecvRequestVote(ths, pMsg, pReply->voteGranted, "", "proceed"); syncLogRecvRequestVote(ths, pMsg, pReply->voteGranted, "", "proceed");
syncLogSendRequestVoteReply(ths, pReply, ""); syncLogSendRequestVoteReply(ths, pReply, "");
(void)syncNodeSendMsgById(&pReply->destId, ths, &rpcMsg); TAOS_CHECK_RETURN(syncNodeSendMsgById(&pReply->destId, ths, &rpcMsg));
if (resetElect) syncNodeResetElectTimer(ths); if (resetElect) syncNodeResetElectTimer(ths);

146
source/libs/wal/src/walMeta.c
View File

@ -57,7 +57,11 @@ static FORCE_INLINE int32_t walScanLogGetLastVer(SWal* pWal, int32_t fileIdx, in
walBuildLogName(pWal, pFileInfo->firstVer, fnameStr); walBuildLogName(pWal, pFileInfo->firstVer, fnameStr);
int64_t fileSize = 0; int64_t fileSize = 0;
(void)taosStatFile(fnameStr, &fileSize, NULL, NULL); if (taosStatFile(fnameStr, &fileSize, NULL, NULL) != 0) {
wError("vgId:%d, failed to stat file due to %s. file:%s", pWal->cfg.vgId, strerror(errno), fnameStr);
code = terrno;
goto _err;
}
TdFilePtr pFile = taosOpenFile(fnameStr, TD_FILE_READ | TD_FILE_WRITE); TdFilePtr pFile = taosOpenFile(fnameStr, TD_FILE_READ | TD_FILE_WRITE);
if (pFile == NULL) { if (pFile == NULL) {
@ -356,6 +360,7 @@ static int32_t walLogEntriesComplete(const SWal* pWal) {
} }
static int32_t walTrimIdxFile(SWal* pWal, int32_t fileIdx) { static int32_t walTrimIdxFile(SWal* pWal, int32_t fileIdx) {
int32_t code = 0;
SWalFileInfo* pFileInfo = taosArrayGet(pWal->fileInfoSet, fileIdx); SWalFileInfo* pFileInfo = taosArrayGet(pWal->fileInfoSet, fileIdx);
if (!pFileInfo) { if (!pFileInfo) {
TAOS_RETURN(TSDB_CODE_FAILED); TAOS_RETURN(TSDB_CODE_FAILED);
@ -365,7 +370,11 @@ static int32_t walTrimIdxFile(SWal* pWal, int32_t fileIdx) {
walBuildIdxName(pWal, pFileInfo->firstVer, fnameStr); walBuildIdxName(pWal, pFileInfo->firstVer, fnameStr);
int64_t fileSize = 0; int64_t fileSize = 0;
(void)taosStatFile(fnameStr, &fileSize, NULL, NULL); if (taosStatFile(fnameStr, &fileSize, NULL, NULL) != 0) {
wError("vgId:%d, failed to stat file due to %s. file:%s", pWal->cfg.vgId, strerror(errno), fnameStr);
code = terrno;
TAOS_RETURN(code);
}
int64_t records = TMAX(0, pFileInfo->lastVer - pFileInfo->firstVer + 1); int64_t records = TMAX(0, pFileInfo->lastVer - pFileInfo->firstVer + 1);
int64_t lastEndOffset = records * sizeof(SWalIdxEntry); int64_t lastEndOffset = records * sizeof(SWalIdxEntry);
@ -381,7 +390,11 @@ static int32_t walTrimIdxFile(SWal* pWal, int32_t fileIdx) {
wInfo("vgId:%d, trim idx file. file: %s, size: %" PRId64 ", offset: %" PRId64, pWal->cfg.vgId, fnameStr, fileSize, wInfo("vgId:%d, trim idx file. file: %s, size: %" PRId64 ", offset: %" PRId64, pWal->cfg.vgId, fnameStr, fileSize,
lastEndOffset); lastEndOffset);
(void)taosFtruncateFile(pFile, lastEndOffset); code = taosFtruncateFile(pFile, lastEndOffset);
if (code < 0) {
wError("vgId:%d, failed to truncate file due to %s. file:%s", pWal->cfg.vgId, strerror(errno), fnameStr);
TAOS_RETURN(code);
}
(void)taosCloseFile(&pFile); (void)taosCloseFile(&pFile);
TAOS_RETURN(TSDB_CODE_SUCCESS); TAOS_RETURN(TSDB_CODE_SUCCESS);
@ -395,8 +408,14 @@ int32_t walCheckAndRepairMeta(SWal* pWal) {
regex_t logRegPattern; regex_t logRegPattern;
regex_t idxRegPattern; regex_t idxRegPattern;
(void)regcomp(&logRegPattern, logPattern, REG_EXTENDED); if (regcomp(&logRegPattern, logPattern, REG_EXTENDED) != 0) {
(void)regcomp(&idxRegPattern, idxPattern, REG_EXTENDED); wError("failed to compile log pattern, error:%s", tstrerror(terrno));
return terrno;
}
if (regcomp(&idxRegPattern, idxPattern, REG_EXTENDED) != 0) {
wError("failed to compile idx pattern");
return terrno;
}
TdDirPtr pDir = taosOpenDir(pWal->path); TdDirPtr pDir = taosOpenDir(pWal->path);
if (pDir == NULL) { if (pDir == NULL) {
@ -420,14 +439,22 @@ int32_t walCheckAndRepairMeta(SWal* pWal) {
if (!taosArrayPush(actualLog, &fileInfo)) { if (!taosArrayPush(actualLog, &fileInfo)) {
regfree(&logRegPattern); regfree(&logRegPattern);
regfree(&idxRegPattern); regfree(&idxRegPattern);
(void)taosCloseDir(&pDir); int32_t ret = taosCloseDir(&pDir);
if (ret != 0) {
wError("failed to close dir, ret:%s", tstrerror(ret));
return terrno;
}
return terrno; return terrno;
} }
} }
} }
(void)taosCloseDir(&pDir); int32_t ret = taosCloseDir(&pDir);
if (ret != 0) {
wError("failed to close dir, ret:%s", tstrerror(ret));
return terrno;
}
regfree(&logRegPattern); regfree(&logRegPattern);
regfree(&idxRegPattern); regfree(&idxRegPattern);
@ -684,7 +711,9 @@ _err:
int64_t walGetVerRetention(SWal* pWal, int64_t bytes) { int64_t walGetVerRetention(SWal* pWal, int64_t bytes) {
int64_t ver = -1; int64_t ver = -1;
int64_t totSize = 0; int64_t totSize = 0;
(void)taosThreadRwlockRdlock(&pWal->mutex); if (taosThreadRwlockRdlock(&pWal->mutex) != 0) {
wError("vgId:%d failed to lock %p", pWal->cfg.vgId, &pWal->mutex);
}
int32_t fileIdx = taosArrayGetSize(pWal->fileInfoSet); int32_t fileIdx = taosArrayGetSize(pWal->fileInfoSet);
while (--fileIdx) { while (--fileIdx) {
SWalFileInfo* pInfo = taosArrayGet(pWal->fileInfoSet, fileIdx); SWalFileInfo* pInfo = taosArrayGet(pWal->fileInfoSet, fileIdx);
@ -694,7 +723,9 @@ int64_t walGetVerRetention(SWal* pWal, int64_t bytes) {
} }
totSize += pInfo->fileSize; totSize += pInfo->fileSize;
} }
(void)taosThreadRwlockUnlock(&pWal->mutex); if (taosThreadRwlockUnlock(&pWal->mutex) != 0) {
wError("vgId:%d failed to lock %p", pWal->cfg.vgId, &pWal->mutex);
}
return ver + 1; return ver + 1;
} }
@ -765,21 +796,35 @@ int32_t walMetaSerialize(SWal* pWal, char** serialized) {
TAOS_RETURN(TSDB_CODE_OUT_OF_MEMORY); TAOS_RETURN(TSDB_CODE_OUT_OF_MEMORY);
} }
(void)cJSON_AddItemToObject(pRoot, "meta", pMeta); if (cJSON_AddItemToObject(pRoot, "meta", pMeta) != 0) {
wInfo("vgId:%d, failed to add meta to root", pWal->cfg.vgId);
}
(void)sprintf(buf, "%" PRId64, pWal->vers.firstVer); (void)sprintf(buf, "%" PRId64, pWal->vers.firstVer);
(void)cJSON_AddStringToObject(pMeta, "firstVer", buf); if (cJSON_AddStringToObject(pMeta, "firstVer", buf) == NULL) {
wInfo("vgId:%d, failed to add firstVer to meta", pWal->cfg.vgId);
}
(void)sprintf(buf, "%" PRId64, pWal->vers.snapshotVer); (void)sprintf(buf, "%" PRId64, pWal->vers.snapshotVer);
(void)cJSON_AddStringToObject(pMeta, "snapshotVer", buf); if (cJSON_AddStringToObject(pMeta, "snapshotVer", buf) == NULL) {
wInfo("vgId:%d, failed to add snapshotVer to meta", pWal->cfg.vgId);
}
(void)sprintf(buf, "%" PRId64, pWal->vers.commitVer); (void)sprintf(buf, "%" PRId64, pWal->vers.commitVer);
(void)cJSON_AddStringToObject(pMeta, "commitVer", buf); if (cJSON_AddStringToObject(pMeta, "commitVer", buf) == NULL) {
wInfo("vgId:%d, failed to add commitVer to meta", pWal->cfg.vgId);
}
(void)sprintf(buf, "%" PRId64, pWal->vers.lastVer); (void)sprintf(buf, "%" PRId64, pWal->vers.lastVer);
(void)cJSON_AddStringToObject(pMeta, "lastVer", buf); if (cJSON_AddStringToObject(pMeta, "lastVer", buf) == NULL) {
wInfo("vgId:%d, failed to add lastVer to meta", pWal->cfg.vgId);
}
(void)cJSON_AddItemToObject(pRoot, "files", pFiles); if (cJSON_AddItemToObject(pRoot, "files", pFiles) != 0) {
wInfo("vgId:%d, failed to add files to root", pWal->cfg.vgId);
}
SWalFileInfo* pData = pWal->fileInfoSet->pData; SWalFileInfo* pData = pWal->fileInfoSet->pData;
for (int i = 0; i < sz; i++) { for (int i = 0; i < sz; i++) {
SWalFileInfo* pInfo = &pData[i]; SWalFileInfo* pInfo = &pData[i];
(void)cJSON_AddItemToArray(pFiles, pField = cJSON_CreateObject()); if (cJSON_AddItemToArray(pFiles, pField = cJSON_CreateObject()) != 0) {
wInfo("vgId:%d, failed to add field to files", pWal->cfg.vgId);
}
if (pField == NULL) { if (pField == NULL) {
cJSON_Delete(pRoot); cJSON_Delete(pRoot);
@ -788,15 +833,25 @@ int32_t walMetaSerialize(SWal* pWal, char** serialized) {
// cjson only support int32_t or double // cjson only support int32_t or double
// string are used to prohibit the loss of precision // string are used to prohibit the loss of precision
(void)sprintf(buf, "%" PRId64, pInfo->firstVer); (void)sprintf(buf, "%" PRId64, pInfo->firstVer);
(void)cJSON_AddStringToObject(pField, "firstVer", buf); if (cJSON_AddStringToObject(pField, "firstVer", buf) == NULL) {
wInfo("vgId:%d, failed to add firstVer to field", pWal->cfg.vgId);
}
(void)sprintf(buf, "%" PRId64, pInfo->lastVer); (void)sprintf(buf, "%" PRId64, pInfo->lastVer);
(void)cJSON_AddStringToObject(pField, "lastVer", buf); if (cJSON_AddStringToObject(pField, "lastVer", buf) == NULL) {
wInfo("vgId:%d, failed to add lastVer to field", pWal->cfg.vgId);
}
(void)sprintf(buf, "%" PRId64, pInfo->createTs); (void)sprintf(buf, "%" PRId64, pInfo->createTs);
(void)cJSON_AddStringToObject(pField, "createTs", buf); if (cJSON_AddStringToObject(pField, "createTs", buf) == NULL) {
wInfo("vgId:%d, failed to add createTs to field", pWal->cfg.vgId);
}
(void)sprintf(buf, "%" PRId64, pInfo->closeTs); (void)sprintf(buf, "%" PRId64, pInfo->closeTs);
(void)cJSON_AddStringToObject(pField, "closeTs", buf); if (cJSON_AddStringToObject(pField, "closeTs", buf) == NULL) {
wInfo("vgId:%d, failed to add closeTs to field", pWal->cfg.vgId);
}
(void)sprintf(buf, "%" PRId64, pInfo->fileSize); (void)sprintf(buf, "%" PRId64, pInfo->fileSize);
(void)cJSON_AddStringToObject(pField, "fileSize", buf); if (cJSON_AddStringToObject(pField, "fileSize", buf) == NULL) {
wInfo("vgId:%d, failed to add fileSize to field", pWal->cfg.vgId);
}
} }
char* pSerialized = cJSON_Print(pRoot); char* pSerialized = cJSON_Print(pRoot);
cJSON_Delete(pRoot); cJSON_Delete(pRoot);
@ -874,7 +929,10 @@ _err:
static int walFindCurMetaVer(SWal* pWal) { static int walFindCurMetaVer(SWal* pWal) {
const char* pattern = "^meta-ver[0-9]+$"; const char* pattern = "^meta-ver[0-9]+$";
regex_t walMetaRegexPattern; regex_t walMetaRegexPattern;
(void)regcomp(&walMetaRegexPattern, pattern, REG_EXTENDED); if (regcomp(&walMetaRegexPattern, pattern, REG_EXTENDED) != 0) {
wError("failed to compile wal meta pattern, error %s", tstrerror(terrno));
return terrno;
}
TdDirPtr pDir = taosOpenDir(pWal->path); TdDirPtr pDir = taosOpenDir(pWal->path);
if (pDir == NULL) { if (pDir == NULL) {
@ -896,7 +954,10 @@ static int walFindCurMetaVer(SWal* pWal) {
} }
wDebug("vgId:%d, wal find current meta: %s is not meta file", pWal->cfg.vgId, name); wDebug("vgId:%d, wal find current meta: %s is not meta file", pWal->cfg.vgId, name);
} }
(void)taosCloseDir(&pDir); if (taosCloseDir(&pDir) != 0) {
wError("failed to close dir, ret:%s", tstrerror(terrno));
return terrno;
}
regfree(&walMetaRegexPattern); regfree(&walMetaRegexPattern);
return metaVer; return metaVer;
} }
@ -979,21 +1040,31 @@ int32_t walSaveMeta(SWal* pWal) {
// delete old file // delete old file
if (metaVer > -1) { if (metaVer > -1) {
(void)walBuildMetaName(pWal, metaVer, fnameStr); n = walBuildMetaName(pWal, metaVer, fnameStr);
(void)taosRemoveFile(fnameStr); if (n >= sizeof(fnameStr)) {
TAOS_RETURN(TAOS_SYSTEM_ERROR(errno));
}
code = taosRemoveFile(fnameStr);
if (code) {
wError("vgId:%d, failed to remove file due to %s. file:%s", pWal->cfg.vgId, strerror(errno), fnameStr);
} else {
wInfo("vgId:%d, remove old meta file: %s", pWal->cfg.vgId, fnameStr);
}
} }
taosMemoryFree(serialized); taosMemoryFree(serialized);
return code; return code;
_err: _err:
taosCloseFile(&pMetaFile); wError("vgId:%d, %s failed at line %d since %s", pWal->cfg.vgId, __func__, lino, tstrerror(code));
(void)taosCloseFile(&pMetaFile);
taosMemoryFree(serialized); taosMemoryFree(serialized);
return code; return code;
} }
int32_t walLoadMeta(SWal* pWal) { int32_t walLoadMeta(SWal* pWal) {
int32_t code = 0; int32_t code = 0;
int n = 0;
// find existing meta file // find existing meta file
int metaVer = walFindCurMetaVer(pWal); int metaVer = walFindCurMetaVer(pWal);
if (metaVer == -1) { if (metaVer == -1) {
@ -1002,12 +1073,24 @@ int32_t walLoadMeta(SWal* pWal) {
TAOS_RETURN(TSDB_CODE_FAILED); TAOS_RETURN(TSDB_CODE_FAILED);
} }
char fnameStr[WAL_FILE_LEN]; char fnameStr[WAL_FILE_LEN];
(void)walBuildMetaName(pWal, metaVer, fnameStr); n = walBuildMetaName(pWal, metaVer, fnameStr);
if (n >= sizeof(fnameStr)) {
TAOS_RETURN(TAOS_SYSTEM_ERROR(errno));
}
// read metafile // read metafile
int64_t fileSize = 0; int64_t fileSize = 0;
(void)taosStatFile(fnameStr, &fileSize, NULL, NULL); if (taosStatFile(fnameStr, &fileSize, NULL, NULL) != 0) {
wError("vgId:%d, failed to stat file due to %s. file:%s", pWal->cfg.vgId, strerror(errno), fnameStr);
code = terrno;
TAOS_RETURN(code);
}
if (fileSize == 0) { if (fileSize == 0) {
(void)taosRemoveFile(fnameStr); code = taosRemoveFile(fnameStr);
if (code) {
wError("vgId:%d, failed to remove file due to %s. file:%s", pWal->cfg.vgId, strerror(errno), fnameStr);
} else {
wInfo("vgId:%d, remove old meta file: %s", pWal->cfg.vgId, fnameStr);
}
wDebug("vgId:%d, wal find empty meta ver %d", pWal->cfg.vgId, metaVer); wDebug("vgId:%d, wal find empty meta ver %d", pWal->cfg.vgId, metaVer);
TAOS_RETURN(TSDB_CODE_FAILED); TAOS_RETURN(TSDB_CODE_FAILED);
@ -1046,6 +1129,9 @@ int32_t walRemoveMeta(SWal* pWal) {
int metaVer = walFindCurMetaVer(pWal); int metaVer = walFindCurMetaVer(pWal);
if (metaVer == -1) return 0; if (metaVer == -1) return 0;
char fnameStr[WAL_FILE_LEN]; char fnameStr[WAL_FILE_LEN];
(void)walBuildMetaName(pWal, metaVer, fnameStr); int n = walBuildMetaName(pWal, metaVer, fnameStr);
if (n >= sizeof(fnameStr)) {
TAOS_RETURN(TAOS_SYSTEM_ERROR(errno));
}
return taosRemoveFile(fnameStr); return taosRemoveFile(fnameStr);
} }

20
source/libs/wal/src/walMgmt.c
View File

@ -160,7 +160,9 @@ SWal *walOpen(const char *path, SWalCfg *pCfg) {
pWal->writeHead.magic = WAL_MAGIC; pWal->writeHead.magic = WAL_MAGIC;
// load meta // load meta
(void)walLoadMeta(pWal); if (walLoadMeta(pWal) < 0) {
wInfo("vgId:%d, failed to load meta since %s", pWal->cfg.vgId, tstrerror(terrno));
}
if (walCheckAndRepairMeta(pWal) < 0) { if (walCheckAndRepairMeta(pWal) < 0) {
wError("vgId:%d, cannot open wal since repair meta file failed", pWal->cfg.vgId); wError("vgId:%d, cannot open wal since repair meta file failed", pWal->cfg.vgId);
@ -233,7 +235,9 @@ int32_t walPersist(SWal *pWal) {
void walClose(SWal *pWal) { void walClose(SWal *pWal) {
TAOS_UNUSED(taosThreadRwlockWrlock(&pWal->mutex)); TAOS_UNUSED(taosThreadRwlockWrlock(&pWal->mutex));
(void)walSaveMeta(pWal); if (walSaveMeta(pWal) < 0) {
wError("vgId:%d, failed to save meta since %s", pWal->cfg.vgId, tstrerror(terrno));
}
TAOS_UNUSED(taosCloseFile(&pWal->pLogFile)); TAOS_UNUSED(taosCloseFile(&pWal->pLogFile));
pWal->pLogFile = NULL; pWal->pLogFile = NULL;
(void)taosCloseFile(&pWal->pIdxFile); (void)taosCloseFile(&pWal->pIdxFile);
@ -257,10 +261,14 @@ void walClose(SWal *pWal) {
if (pWal->cfg.level == TAOS_WAL_SKIP) { if (pWal->cfg.level == TAOS_WAL_SKIP) {
wInfo("vgId:%d, remove all wals, path:%s", pWal->cfg.vgId, pWal->path); wInfo("vgId:%d, remove all wals, path:%s", pWal->cfg.vgId, pWal->path);
taosRemoveDir(pWal->path); taosRemoveDir(pWal->path);
(void)taosMkDir(pWal->path); if (taosMkDir(pWal->path) != 0) {
wError("vgId:%d, path:%s, failed to create directory since %s", pWal->cfg.vgId, pWal->path, tstrerror(terrno));
}
} }
(void)taosRemoveRef(tsWal.refSetId, pWal->refId); if (taosRemoveRef(tsWal.refSetId, pWal->refId) < 0) {
wError("vgId:%d, failed to remove ref for Wal since %s", pWal->cfg.vgId, tstrerror(terrno));
}
} }
static void walFreeObj(void *wal) { static void walFreeObj(void *wal) {
@ -285,7 +293,9 @@ static bool walNeedFsync(SWal *pWal) {
static void walUpdateSeq() { static void walUpdateSeq() {
taosMsleep(WAL_REFRESH_MS); taosMsleep(WAL_REFRESH_MS);
(void)atomic_add_fetch_32((volatile int32_t *)&tsWal.seq, 1); if (atomic_add_fetch_32((volatile int32_t *)&tsWal.seq, 1) < 0) {
wError("failed to update wal seq since %s", strerror(errno));
}
} }
static void walFsyncAll() { static void walFsyncAll() {

7
source/libs/wal/src/walRef.c
View File

@ -52,8 +52,11 @@ void walCloseRef(SWal *pWal, int64_t refId) {
} else { } else {
wDebug("vgId:%d, wal close ref null, refId %" PRId64, pWal->cfg.vgId, refId); wDebug("vgId:%d, wal close ref null, refId %" PRId64, pWal->cfg.vgId, refId);
} }
int32_t code = 0;
(void)taosHashRemove(pWal->pRefHash, &refId, sizeof(int64_t)); code = taosHashRemove(pWal->pRefHash, &refId, sizeof(int64_t));
if (code) {
wError("vgId:%d, wal remove ref failed, refId %" PRId64 ", error:%s", pWal->cfg.vgId, refId, tstrerror(code));
}
} }
} }

5
source/util/src/tlrucache.c
View File

@ -14,12 +14,12 @@
*/ */
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "tlrucache.h"
#include "os.h" #include "os.h"
#include "taoserror.h" #include "taoserror.h"
#include "tarray.h" #include "tarray.h"
#include "tdef.h" #include "tdef.h"
#include "tlog.h" #include "tlog.h"
#include "tlrucache.h"
#include "tutil.h" #include "tutil.h"
typedef struct SLRUEntry SLRUEntry; typedef struct SLRUEntry SLRUEntry;
@ -306,7 +306,6 @@ static void taosLRUCacheShardEvictLRU(SLRUCacheShard *shard, size_t charge, SArr
taosLRUCacheShardLRURemove(shard, old); taosLRUCacheShardLRURemove(shard, old);
SLRUEntry *tentry = taosLRUEntryTableRemove(&shard->table, old->keyData, old->keyLength, old->hash); SLRUEntry *tentry = taosLRUEntryTableRemove(&shard->table, old->keyData, old->keyLength, old->hash);
TAOS_LRU_ENTRY_SET_IN_CACHE(old, false); TAOS_LRU_ENTRY_SET_IN_CACHE(old, false);
shard->usage -= old->totalCharge; shard->usage -= old->totalCharge;
@ -870,4 +869,4 @@ bool taosLRUCacheIsStrictCapacity(SLRUCache *cache) {
(void)taosThreadMutexUnlock(&cache->shardedCache.capacityMutex); (void)taosThreadMutexUnlock(&cache->shardedCache.capacityMutex);
return strict; return strict;
} }

6
tests/parallel_test/run_case.sh
View File

@ -76,15 +76,15 @@ ulimit -c unlimited
md5sum /usr/lib/libtaos.so.1 md5sum /usr/lib/libtaos.so.1
md5sum /home/TDinternal/debug/build/lib/libtaos.so md5sum /home/TDinternal/debug/build/lib/libtaos.so
#define taospy 2.7.10 #define taospy 2.7.16
pip3 list|grep taospy pip3 list|grep taospy
pip3 uninstall taospy -y pip3 uninstall taospy -y
pip3 install --default-timeout=120 taospy==2.7.15 pip3 install --default-timeout=120 taospy==2.7.16
#define taos-ws-py 0.3.1 #define taos-ws-py 0.3.1
pip3 list|grep taos-ws-py pip3 list|grep taos-ws-py
pip3 uninstall taos-ws-py -y pip3 uninstall taos-ws-py -y
pip3 install --default-timeout=600 taos-ws-py==0.3.1 pip3 install --default-timeout=600 taos-ws-py==0.3.3
$TIMEOUT_CMD $cmd $TIMEOUT_CMD $cmd
RET=$? RET=$?