diff --git a/src/common/src/tglobal.c b/src/common/src/tglobal.c index 4b5392257ffa072c53c4332733518c8f1f68f8bd..17eb5714e0272bcebe21de84d938f94708b7c4d9 100644 --- a/src/common/src/tglobal.c +++ b/src/common/src/tglobal.c @@ -206,7 +206,7 @@ int32_t tsNumOfLogLines = 10000000; int32_t mDebugFlag = 131; int32_t sdbDebugFlag = 131; int32_t dDebugFlag = 135; -int32_t vDebugFlag = 131; +int32_t vDebugFlag = 135; int32_t cDebugFlag = 131; int32_t jniDebugFlag = 131; int32_t odbcDebugFlag = 131; diff --git a/src/cq/src/cqMain.c b/src/cq/src/cqMain.c index 3968d5b8c9dfbb53ad04179e0451fc9aab76980e..e278c3a7ccd145585b70d3488abaf1a9dceab571 100644 --- a/src/cq/src/cqMain.c +++ b/src/cq/src/cqMain.c @@ -161,7 +161,7 @@ void cqStop(void *handle) { return; } SCqContext *pContext = handle; - cInfo("vgId:%d, stop all CQs", pContext->vgId); + cDebug("vgId:%d, stop all CQs", pContext->vgId); if (pContext->dbConn == NULL || pContext->master == 0) return; pthread_mutex_lock(&pContext->mutex); diff --git a/src/inc/taoserror.h b/src/inc/taoserror.h index e0d7e018434f15d96fe559440ba6b07f027cd930..be33262f7f75f20485c977c3b3a7378a6021481a 100644 --- a/src/inc/taoserror.h +++ b/src/inc/taoserror.h @@ -206,9 +206,10 @@ TAOS_DEFINE_ERROR(TSDB_CODE_VND_NO_SUCH_FILE_OR_DIR, 0, 0x0507, "Missing da TAOS_DEFINE_ERROR(TSDB_CODE_VND_OUT_OF_MEMORY, 0, 0x0508, "Out of memory") TAOS_DEFINE_ERROR(TSDB_CODE_VND_APP_ERROR, 0, 0x0509, "Unexpected generic error in vnode") TAOS_DEFINE_ERROR(TSDB_CODE_VND_INVALID_VRESION_FILE, 0, 0x050A, "Invalid version file") -TAOS_DEFINE_ERROR(TSDB_CODE_VND_IS_FULL, 0, 0x050B, "Vnode memory is full because commit failed") +TAOS_DEFINE_ERROR(TSDB_CODE_VND_IS_FULL, 0, 0x050B, "Database memory is full for commit failed") +TAOS_DEFINE_ERROR(TSDB_CODE_VND_IS_FLOWCTRL, 0, 0x050C, "Database memory is full for waiting commit") TAOS_DEFINE_ERROR(TSDB_CODE_VND_NOT_SYNCED, 0, 0x0511, "Database suspended") -TAOS_DEFINE_ERROR(TSDB_CODE_VND_NO_WRITE_AUTH, 0, 0x0512, "Write operation denied") +TAOS_DEFINE_ERROR(TSDB_CODE_VND_NO_WRITE_AUTH, 0, 0x0512, "Database write operation denied") TAOS_DEFINE_ERROR(TSDB_CODE_VND_SYNCING, 0, 0x0513, "Database is syncing") // tsdb diff --git a/src/mnode/src/mnodeProfile.c b/src/mnode/src/mnodeProfile.c index 3256d5cd59699139cf67cebcc41070ca99cffad6..5d63ae9ff411cdddd82400d1f2853e6e882ba527 100644 --- a/src/mnode/src/mnodeProfile.c +++ b/src/mnode/src/mnodeProfile.c @@ -282,25 +282,32 @@ static int32_t mnodeRetrieveConns(SShowObj *pShow, char *data, int32_t rows, voi // not thread safe, need optimized int32_t mnodeSaveQueryStreamList(SConnObj *pConn, SHeartBeatMsg *pHBMsg) { - pConn->numOfQueries = htonl(pHBMsg->numOfQueries); - if (pConn->numOfQueries > 0) { + pConn->numOfQueries = 0; + pConn->numOfStreams = 0; + + int32_t numOfQueries = htonl(pHBMsg->numOfQueries); + if (numOfQueries > 0) { if (pConn->pQueries == NULL) { pConn->pQueries = calloc(sizeof(SQueryDesc), QUERY_STREAM_SAVE_SIZE); } - int32_t saveSize = MIN(QUERY_STREAM_SAVE_SIZE, pConn->numOfQueries) * sizeof(SQueryDesc); + pConn->numOfQueries = MIN(QUERY_STREAM_SAVE_SIZE, numOfQueries); + + int32_t saveSize = pConn->numOfQueries * sizeof(SQueryDesc); if (saveSize > 0 && pConn->pQueries != NULL) { memcpy(pConn->pQueries, pHBMsg->pData, saveSize); } } - pConn->numOfStreams = htonl(pHBMsg->numOfStreams); - if (pConn->numOfStreams > 0) { + int32_t numOfStreams = htonl(pHBMsg->numOfStreams); + if (numOfStreams > 0) { if (pConn->pStreams == NULL) { pConn->pStreams = calloc(sizeof(SStreamDesc), QUERY_STREAM_SAVE_SIZE); } - int32_t saveSize = MIN(QUERY_STREAM_SAVE_SIZE, pConn->numOfStreams) * sizeof(SStreamDesc); + pConn->numOfStreams = MIN(QUERY_STREAM_SAVE_SIZE, numOfStreams); + + int32_t saveSize = pConn->numOfStreams * sizeof(SStreamDesc); if (saveSize > 0 && pConn->pStreams != NULL) { memcpy(pConn->pStreams, pHBMsg->pData + pConn->numOfQueries * sizeof(SQueryDesc), saveSize); } diff --git a/src/sync/src/syncMain.c b/src/sync/src/syncMain.c index c86265d55622fc563b781aae851ee5af8f35b752..eea0ba4bb18eb1e1e93b1a1c1fb951405bdcd8a2 100644 --- a/src/sync/src/syncMain.c +++ b/src/sync/src/syncMain.c @@ -1312,7 +1312,7 @@ static int32_t syncForwardToPeerImpl(SSyncNode *pNode, void *data, void *mhandle } // always update version - sTrace("vgId:%d, forward to peer, replica:%d role:%s qtype:%s hver:%" PRIu64, pNode->vgId, pNode->replica, + sTrace("vgId:%d, update nodeVersion, replica:%d role:%s qtype:%s hver:%" PRIu64, pNode->vgId, pNode->replica, syncRole[nodeRole], qtypeStr[qtype], pWalHead->version); nodeVersion = pWalHead->version; diff --git a/src/sync/src/syncRestore.c b/src/sync/src/syncRestore.c index d156c93865a3f4757cd9c9af0aff7b6789f59822..3d262d6e7fa6d092d760df87b7b0e76bbd2e1dba 100644 --- a/src/sync/src/syncRestore.c +++ b/src/sync/src/syncRestore.c @@ -43,7 +43,7 @@ static void syncRemoveExtraFile(SSyncPeer *pPeer, int32_t sindex, int32_t eindex snprintf(fname, sizeof(fname), "%s/%s", pNode->path, name); (void)remove(fname); - sDebug("%s, %s is removed", pPeer->id, fname); + sInfo("%s, %s is removed for its extra", pPeer->id, fname); index++; if (index > eindex) break; diff --git a/src/vnode/inc/vnodeInt.h b/src/vnode/inc/vnodeInt.h index 34f7d64ed11ea509d0a87ab07108d2bd3ee5b318..b28eb690fec7656e696498fc2dcffef8076fd47c 100644 --- a/src/vnode/inc/vnodeInt.h +++ b/src/vnode/inc/vnodeInt.h @@ -46,6 +46,7 @@ typedef struct { int8_t isFull; int8_t isCommiting; uint64_t version; // current version + uint64_t cversion; // version while commit start uint64_t fversion; // version on saved data file void * wqueue; // write queue void * qqueue; // read query queue diff --git a/src/vnode/src/vnodeMain.c b/src/vnode/src/vnodeMain.c index 3a603466f4c02b01624eda462986e917f93917ea..425c333c8a2079c1ec16a3052b00aab7a4cb99a1 100644 --- a/src/vnode/src/vnodeMain.c +++ b/src/vnode/src/vnodeMain.c @@ -203,8 +203,8 @@ int32_t vnodeOpen(int32_t vgId) { code = vnodeReadVersion(pVnode); if (code != TSDB_CODE_SUCCESS) { - vError("vgId:%d, failed to read version, generate it from data file", pVnode->vgId); - // Allow vnode start even when read version fails, set version as walVersion or zero + vError("vgId:%d, failed to read file version, generate it from data file", pVnode->vgId); + // Allow vnode start even when read file version fails, set file version as wal version or zero // vnodeCleanUp(pVnode); // return code; } @@ -442,6 +442,7 @@ static int32_t vnodeProcessTsdbStatus(void *arg, int32_t status, int32_t eno) { pVnode->fversion, pVnode->version); pVnode->isCommiting = 0; pVnode->isFull = 1; + pVnode->cversion = pVnode->version; return 0; } @@ -457,7 +458,7 @@ static int32_t vnodeProcessTsdbStatus(void *arg, int32_t status, int32_t eno) { if (status == TSDB_STATUS_COMMIT_OVER) { pVnode->isCommiting = 0; pVnode->isFull = 0; - pVnode->fversion = pVnode->version; + pVnode->fversion = pVnode->cversion; vDebug("vgId:%d, commit over, fver:%" PRIu64 " vver:%" PRIu64, pVnode->vgId, pVnode->fversion, pVnode->version); if (!vnodeInInitStatus(pVnode)) { walRemoveOneOldFile(pVnode->wal); diff --git a/src/vnode/src/vnodeRead.c b/src/vnode/src/vnodeRead.c index 03d1272771d599e9115896148007c9b1fe310151..7ad97396c2df2b105056861e5c2916de92c7d8d9 100644 --- a/src/vnode/src/vnodeRead.c +++ b/src/vnode/src/vnodeRead.c @@ -133,7 +133,7 @@ static int32_t vnodePutItemIntoReadQueue(SVnodeObj *pVnode, void **qhandle, void int32_t code = vnodeWriteToRQueue(pVnode, qhandle, 0, TAOS_QTYPE_QUERY, &rpcMsg); if (code == TSDB_CODE_SUCCESS) { - vDebug("QInfo:%p add to vread queue for exec query", *qhandle); + vTrace("QInfo:%p add to vread queue for exec query", *qhandle); } return code; @@ -164,7 +164,7 @@ static int32_t vnodeDumpQueryResult(SRspRet *pRet, void *pVnode, void **handle, } } else { *freeHandle = true; - vDebug("QInfo:%p exec completed, free handle:%d", *handle, *freeHandle); + vTrace("QInfo:%p exec completed, free handle:%d", *handle, *freeHandle); } } else { SRetrieveTableRsp *pRsp = (SRetrieveTableRsp *)rpcMallocCont(sizeof(SRetrieveTableRsp)); @@ -266,7 +266,7 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pRead) { } if (handle != NULL) { - vDebug("vgId:%d, QInfo:%p, dnode query msg disposed, create qhandle and returns to app", vgId, *handle); + vTrace("vgId:%d, QInfo:%p, dnode query msg disposed, create qhandle and returns to app", vgId, *handle); code = vnodePutItemIntoReadQueue(pVnode, handle, pRead->rpcHandle); if (code != TSDB_CODE_SUCCESS) { pRsp->code = code; @@ -278,7 +278,7 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pRead) { assert(pCont != NULL); void **qhandle = (void **)pRead->qhandle; - vDebug("vgId:%d, QInfo:%p, dnode continues to exec query", pVnode->vgId, *qhandle); + vTrace("vgId:%d, QInfo:%p, dnode continues to exec query", pVnode->vgId, *qhandle); // In the retrieve blocking model, only 50% CPU will be used in query processing if (tsHalfCoresForQuery) { @@ -294,7 +294,7 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pRead) { pRead->rpcHandle = qGetResultRetrieveMsg(*qhandle); assert(pRead->rpcHandle != NULL); - vDebug("vgId:%d, QInfo:%p, start to build retrieval rsp after query paused, %p", pVnode->vgId, *qhandle, + vTrace("vgId:%d, QInfo:%p, start to build retrieval rsp after query paused, %p", pVnode->vgId, *qhandle, pRead->rpcHandle); // set the real rsp error code @@ -327,7 +327,7 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SVReadMsg *pRead) { pRetrieve->free = htons(pRetrieve->free); pRetrieve->qhandle = htobe64(pRetrieve->qhandle); - vDebug("vgId:%d, QInfo:%p, retrieve msg is disposed, free:%d, conn:%p", pVnode->vgId, (void *)pRetrieve->qhandle, + vTrace("vgId:%d, QInfo:%p, retrieve msg is disposed, free:%d, conn:%p", pVnode->vgId, (void *)pRetrieve->qhandle, pRetrieve->free, pRead->rpcHandle); memset(pRet, 0, sizeof(SRspRet)); @@ -410,6 +410,6 @@ int32_t vnodeNotifyCurrentQhandle(void *handle, void *qhandle, int32_t vgId) { pMsg->header.vgId = htonl(vgId); pMsg->header.contLen = htonl(sizeof(SRetrieveTableMsg)); - vDebug("QInfo:%p register qhandle to connect:%p", qhandle, handle); + vTrace("QInfo:%p register qhandle to connect:%p", qhandle, handle); return rpcReportProgress(handle, (char *)pMsg, sizeof(SRetrieveTableMsg)); } diff --git a/src/vnode/src/vnodeWrite.c b/src/vnode/src/vnodeWrite.c index a826a4903fe6a498db0769a1a1d883791bbb3802..7cf1a9059839e2111808b9458b35fb1ad0573f57 100644 --- a/src/vnode/src/vnodeWrite.c +++ b/src/vnode/src/vnodeWrite.c @@ -244,7 +244,7 @@ int32_t vnodeWriteToWQueue(void *vparam, void *wparam, int32_t qtype, void *rpar int32_t queued = atomic_add_fetch_32(&pVnode->queuedWMsg, 1); if (queued > MAX_QUEUED_MSG_NUM) { vDebug("vgId:%d, too many msg:%d in vwqueue, flow control", pVnode->vgId, queued); - taosMsleep(1); + taosMsleep(3); } code = vnodePerformFlowCtrl(pWrite); @@ -271,6 +271,8 @@ static void vnodeFlowCtrlMsgToWQueue(void *param, void *tmrId) { SVnodeObj * pVnode = pWrite->pVnode; int32_t code = TSDB_CODE_VND_SYNCING; + if (pVnode->flowctrlLevel <= 0) code = TSDB_CODE_VND_IS_FLOWCTRL; + pWrite->processedCount++; if (pWrite->processedCount > 100) { vError("vgId:%d, msg:%p, failed to process since %s, retry:%d", pVnode->vgId, pWrite, tstrerror(code), @@ -290,8 +292,8 @@ static void vnodeFlowCtrlMsgToWQueue(void *param, void *tmrId) { static int32_t vnodePerformFlowCtrl(SVWriteMsg *pWrite) { SVnodeObj *pVnode = pWrite->pVnode; - if (pVnode->flowctrlLevel <= 0) return 0; if (pWrite->qtype != TAOS_QTYPE_RPC) return 0; + if (pVnode->queuedWMsg < MAX_QUEUED_MSG_NUM && pVnode->flowctrlLevel <= 0) return 0; if (tsFlowCtrl == 0) { int32_t ms = pow(2, pVnode->flowctrlLevel + 2); diff --git a/src/wal/inc/walInt.h b/src/wal/inc/walInt.h index 06748d885f7b0fb798abc1873a1734abbf9f05ab..890b404ce9489f0f7798aa7ff21b4f1e80da2b04 100644 --- a/src/wal/inc/walInt.h +++ b/src/wal/inc/walInt.h @@ -38,7 +38,7 @@ extern int32_t wDebugFlag; #define WAL_SIGNATURE ((uint32_t)(0xFAFBFDFE)) #define WAL_PATH_LEN (TSDB_FILENAME_LEN + 12) #define WAL_FILE_LEN (WAL_PATH_LEN + 32) -#define WAL_FILE_NUM 3 +#define WAL_FILE_NUM 1 // 3 typedef struct { uint64_t version; diff --git a/src/wal/src/walWrite.c b/src/wal/src/walWrite.c index 9174da3d0397500c3fed11ec8d311ec6b095edc5..2253ad5c33906714352ecdcd0a916dbcc307a82f 100644 --- a/src/wal/src/walWrite.c +++ b/src/wal/src/walWrite.c @@ -173,7 +173,7 @@ int32_t walRestore(void *handle, void *pVnode, FWalWrite writeFp) { continue; } - wInfo("vgId:%d, file:%s, restore success", pWal->vgId, walName); + wInfo("vgId:%d, file:%s, restore success, wver:%" PRIu64, pWal->vgId, walName, pWal->version); count++; } @@ -267,8 +267,6 @@ static int32_t walRestoreWalFile(SWal *pWal, void *pVnode, FWalWrite writeFp, ch return TAOS_SYSTEM_ERROR(errno); } - wDebug("vgId:%d, file:%s, start to restore", pWal->vgId, name); - int32_t code = TSDB_CODE_SUCCESS; int64_t offset = 0; SWalHead *pHead = buffer; diff --git a/tests/script/unique/arbitrator/insert_duplicationTs.sim b/tests/script/unique/arbitrator/insert_duplicationTs.sim index a873bf02ae164e7f3bddd955adaec90568257775..7c6c6e6e92d2ed5d6fc2c2d0ca8796302318ab4e 100644 --- a/tests/script/unique/arbitrator/insert_duplicationTs.sim +++ b/tests/script/unique/arbitrator/insert_duplicationTs.sim @@ -91,8 +91,11 @@ while $i < $tblNum $i = $i + 1 endw +sql show db.vgroups; +print d1: $data04 $data05 , d2: $data06 $data07 + sql select count(*) from $stb -print rows:$rows data00:$data00 +print rtest1==> rows:$rows data00:$data00 if $rows != 1 then return -1 endi @@ -103,6 +106,15 @@ endi $totalRows = $data00 +sql select count(*) from $stb +print test2==> rows:$rows data00:$data00 +sql select count(*) from $stb +print test3==> rows:$rows data00:$data00 +sql select count(*) from $stb +print test4==> rows:$rows data00:$data00 +sql select count(*) from $stb +print test5==> rows:$rows data00:$data00 + print ============== step3: insert old data(now-15d) and new data(now+15d), control data rows in order to save in cache, not falling disc sql insert into $tb values ( now - 20d , -20 ) sql insert into $tb values ( now - 40d , -40 ) @@ -153,12 +165,21 @@ if $data00 != $totalRows then return -1 endi +sql select count(*) from $stb +print data00 $data00 +sql select count(*) from $stb +print data00 $data00 +sql select count(*) from $stb +print data00 $data00 +sql select count(*) from $stb +print data00 $data00 + print ============== step5: insert two data rows: now-16d, now+16d, sql insert into $tb values ( now - 21d , -21 ) sql insert into $tb values ( now - 41d , -41 ) $totalRows = $totalRows + 2 -print ============== step5: restart dnode2, waiting sync end +print ============== step6: restart dnode2, waiting sync end system sh/exec.sh -n dnode2 -s start sleep 3000 $loopCnt = 0 @@ -192,9 +213,81 @@ endi sleep $sleepTimer # check using select +sleep 5000 +sql select count(*) from $stb +print data00 $data00 +if $data00 != $totalRows then + return -1 +endi + +sql select count(*) from $stb +print data00 $data00 +if $data00 != $totalRows then + return -1 +endi + +sql select count(*) from $stb +print data00 $data00 +if $data00 != $totalRows then + return -1 +endi + +sql select count(*) from $stb +print data00 $data00 +if $data00 != $totalRows then + return -1 +endi + +sql select count(*) from $stb +print data00 $data00 +if $data00 != $totalRows then + return -1 +endi + +sql select count(*) from $stb +print data00 $data00 +if $data00 != $totalRows then + return -1 +endi + +sql select count(*) from $stb +print data00 $data00 +if $data00 != $totalRows then + return -1 +endi + sql select count(*) from $stb print data00 $data00 if $data00 != $totalRows then return -1 endi +sql select count(*) from $stb +print data00 $data00 +if $data00 != $totalRows then + return -1 +endi + +sql select count(*) from $stb +print data00 $data00 +if $data00 != $totalRows then + return -1 +endi + +sql select count(*) from $stb +print data00 $data00 +if $data00 != $totalRows then + return -1 +endi + +sql select count(*) from $stb +print data00 $data00 +if $data00 != $totalRows then + return -1 +endi + +sql select count(*) from $stb +print data00 $data00 +if $data00 != $totalRows then + return -1 +endi