diff --git a/src/inc/taoserror.h b/src/inc/taoserror.h index be33262f7f75f20485c977c3b3a7378a6021481a..c913b2cf2a6fb1506d5c6d9b1d483ec063bede01 100644 --- a/src/inc/taoserror.h +++ b/src/inc/taoserror.h @@ -266,6 +266,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_SYN_INVALID_CONFIG, 0, 0x0900, "Invalid Sy TAOS_DEFINE_ERROR(TSDB_CODE_SYN_NOT_ENABLED, 0, 0x0901, "Sync module not enabled") TAOS_DEFINE_ERROR(TSDB_CODE_SYN_INVALID_VERSION, 0, 0x0902, "Invalid Sync version") TAOS_DEFINE_ERROR(TSDB_CODE_SYN_CONFIRM_EXPIRED, 0, 0x0903, "Sync confirm expired") +TAOS_DEFINE_ERROR(TSDB_CODE_SYN_TOO_MANY_FWDINFO, 0, 0x0904, "Too many sync fwd infos") // wal TAOS_DEFINE_ERROR(TSDB_CODE_WAL_APP_ERROR, 0, 0x1000, "Unexpected generic error in wal") diff --git a/src/inc/tsync.h b/src/inc/tsync.h index 1303195ef149a98b6ddd7ffe55615d27eaeba04d..0ce2a1a495b7826635d07ac9139cb77b306c9330 100644 --- a/src/inc/tsync.h +++ b/src/inc/tsync.h @@ -121,7 +121,6 @@ extern char *syncRole[]; //global configurable parameters extern int32_t tsMaxSyncNum; extern int32_t tsSyncTcpThreads; -extern int32_t tsMaxWatchFiles; extern int32_t tsSyncTimer; extern int32_t tsMaxFwdInfo; extern int32_t sDebugFlag; diff --git a/src/sync/src/syncMain.c b/src/sync/src/syncMain.c index 080c7d3514eba702292fa43f368c5ae7ae927e29..aae5dab3cdd5d394307dbbc8c5c4b0040a05d4a7 100644 --- a/src/sync/src/syncMain.c +++ b/src/sync/src/syncMain.c @@ -32,8 +32,7 @@ // global configurable int32_t tsMaxSyncNum = 2; int32_t tsSyncTcpThreads = 2; -int32_t tsMaxWatchFiles = 500; -int32_t tsMaxFwdInfo = 200; +int32_t tsMaxFwdInfo = 512; int32_t tsSyncTimer = 1; // module global, not configurable @@ -60,7 +59,7 @@ static void syncRemoveConfirmedFwdInfo(SSyncNode *pNode); static void syncMonitorFwdInfos(void *param, void *tmrId); static void syncMonitorNodeRole(void *param, void *tmrId); static void syncProcessFwdAck(SSyncNode *pNode, SFwdInfo *pFwdInfo, int32_t code); -static void syncSaveFwdInfo(SSyncNode *pNode, uint64_t version, void *mhandle); +static int32_t syncSaveFwdInfo(SSyncNode *pNode, uint64_t version, void *mhandle); static void syncRestartPeer(SSyncPeer *pPeer); static int32_t syncForwardToPeerImpl(SSyncNode *pNode, void *data, void *mhandle, int32_t qtyp); static SSyncPeer *syncAddPeer(SSyncNode *pNode, const SNodeInfo *pInfo); @@ -892,15 +891,24 @@ static void syncProcessFwdResponse(char *cont, SSyncPeer *pPeer) { sTrace("%s, forward-rsp is received, code:%x hver:%" PRIu64, pPeer->id, pFwdRsp->code, pFwdRsp->version); SFwdInfo *pFirst = pSyncFwds->fwdInfo + pSyncFwds->first; + bool found = false; if (pFirst->version <= pFwdRsp->version && pSyncFwds->fwds > 0) { // find the forwardInfo from first for (int32_t i = 0; i < pSyncFwds->fwds; ++i) { pFwdInfo = pSyncFwds->fwdInfo + (i + pSyncFwds->first) % tsMaxFwdInfo; - if (pFwdRsp->version == pFwdInfo->version) break; + if (pFwdRsp->version == pFwdInfo->version) { + found = true; + syncProcessFwdAck(pNode, pFwdInfo, pFwdRsp->code); + syncRemoveConfirmedFwdInfo(pNode); + break; + } } + } + if (!found) { + sTrace("%s, forward-rsp not found first:%d fwds:%d, code:%x hver:%" PRIu64, pPeer->id, pSyncFwds->first, + pSyncFwds->fwds, pFwdRsp->code, pFwdRsp->version); syncProcessFwdAck(pNode, pFwdInfo, pFwdRsp->code); - syncRemoveConfirmedFwdInfo(pNode); } } @@ -1180,13 +1188,15 @@ static void syncProcessBrokenLink(void *param) { taosReleaseRef(tsSyncRefId, pNode->rid); } -static void syncSaveFwdInfo(SSyncNode *pNode, uint64_t version, void *mhandle) { +static int32_t syncSaveFwdInfo(SSyncNode *pNode, uint64_t version, void *mhandle) { SSyncFwds *pSyncFwds = pNode->pSyncFwds; int64_t time = taosGetTimestampMs(); if (pSyncFwds->fwds >= tsMaxFwdInfo) { - pSyncFwds->first = (pSyncFwds->first + 1) % tsMaxFwdInfo; - pSyncFwds->fwds--; + // pSyncFwds->first = (pSyncFwds->first + 1) % tsMaxFwdInfo; + // pSyncFwds->fwds--; + sError("vgId:%d, failed to save fwd info, hver:%" PRIu64 " fwds:%d", pNode->vgId, version, pSyncFwds->fwds); + return TSDB_CODE_SYN_TOO_MANY_FWDINFO; } if (pSyncFwds->fwds > 0) { @@ -1201,6 +1211,8 @@ static void syncSaveFwdInfo(SSyncNode *pNode, uint64_t version, void *mhandle) { pSyncFwds->fwds++; sTrace("vgId:%d, fwd info is saved, hver:%" PRIu64 " fwds:%d ", pNode->vgId, version, pSyncFwds->fwds); + + return 0; } static void syncRemoveConfirmedFwdInfo(SSyncNode *pNode) { @@ -1214,8 +1226,7 @@ static void syncRemoveConfirmedFwdInfo(SSyncNode *pNode) { pSyncFwds->first = (pSyncFwds->first + 1) % tsMaxFwdInfo; pSyncFwds->fwds--; if (pSyncFwds->fwds == 0) pSyncFwds->first = pSyncFwds->last; - // sDebug("vgId:%d, fwd info is removed, hver:%d, fwds:%d", - // pNode->vgId, pFwdInfo->version, pSyncFwds->fwds); + sTrace("vgId:%d, fwd info is removed, hver:%" PRIu64 " fwds:%d", pNode->vgId, pFwdInfo->version, pSyncFwds->fwds); memset(pFwdInfo, 0, sizeof(SFwdInfo)); } } @@ -1341,8 +1352,8 @@ static int32_t syncForwardToPeerImpl(SSyncNode *pNode, void *data, void *mhandle if (pPeer->role != TAOS_SYNC_ROLE_SLAVE && pPeer->sstatus != TAOS_SYNC_STATUS_CACHE) continue; if (pNode->quorum > 1 && code == 0) { - syncSaveFwdInfo(pNode, pWalHead->version, mhandle); - code = 1; + code = syncSaveFwdInfo(pNode, pWalHead->version, mhandle); + if (code >= 0) code = 1; } int32_t retLen = taosWriteMsg(pPeer->peerFd, pSyncHead, fwdLen); diff --git a/tests/test/c/createTablePerformance.c b/tests/test/c/createTablePerformance.c index eae104291a3bfcbb37e6a2c89d61214c61d4b6c0..b94c687f2cba7310949b0a3b12b6f4fc007e5a9a 100644 --- a/tests/test/c/createTablePerformance.c +++ b/tests/test/c/createTablePerformance.c @@ -32,6 +32,7 @@ int32_t numOfThreads = 30; int32_t numOfTables = 100000; int32_t replica = 1; int32_t numOfColumns = 2; +TAOS * con = NULL; typedef struct { int32_t tableBeginIndex; @@ -84,13 +85,14 @@ int main(int argc, char *argv[]) { pthread_attr_destroy(&thattr); free(pInfo); + taos_close(con); } void createDbAndSTable() { pPrint("start to create db and stable"); char qstr[64000]; - TAOS *con = taos_connect(NULL, "root", "taosdata", NULL, 0); + con = taos_connect(NULL, "root", "taosdata", NULL, 0); if (con == NULL) { pError("failed to connect to DB, reason:%s", taos_errstr(con)); exit(1); @@ -127,8 +129,6 @@ void createDbAndSTable() { exit(0); } taos_free_result(pSql); - - taos_close(con); } void *threadFunc(void *param) { @@ -136,12 +136,6 @@ void *threadFunc(void *param) { char qstr[65000]; int code; - TAOS *con = taos_connect(NULL, "root", "taosdata", NULL, 0); - if (con == NULL) { - pError("index:%d, failed to connect to DB, reason:%s", pInfo->threadIndex, taos_errstr(con)); - exit(1); - } - sprintf(qstr, "use %s", pInfo->dbName); TAOS_RES *pSql = taos_query(con, qstr); taos_free_result(pSql); @@ -170,7 +164,6 @@ void *threadFunc(void *param) { pInfo->createTableSpeed = speed; pPrint("thread:%d, time:%.2f sec, speed:%.1f tables/second, ", pInfo->threadIndex, seconds, speed); - taos_close(con); return 0; }