diff --git a/include/dnode/mnode/sdb/sdb.h b/include/dnode/mnode/sdb/sdb.h index 4fab11e237..a56c6ca16d 100644 --- a/include/dnode/mnode/sdb/sdb.h +++ b/include/dnode/mnode/sdb/sdb.h @@ -107,10 +107,9 @@ typedef enum { typedef enum { SDB_STATUS_INIT = 0, SDB_STATUS_CREATING = 1, - SDB_STATUS_UPDATING = 2, - SDB_STATUS_DROPPING = 3, + SDB_STATUS_DROPPING = 2, + SDB_STATUS_DROPPED = 3, SDB_STATUS_READY = 4, - SDB_STATUS_DROPPED = 5 } ESdbStatus; typedef enum { diff --git a/include/util/taoserror.h b/include/util/taoserror.h index 0b2301739a..00708f0933 100644 --- a/include/util/taoserror.h +++ b/include/util/taoserror.h @@ -87,6 +87,7 @@ int32_t* taosGetErrno(); #define TSDB_CODE_MSG_NOT_PROCESSED TAOS_DEF_ERROR_CODE(0, 0x0113) #define TSDB_CODE_CFG_NOT_FOUND TAOS_DEF_ERROR_CODE(0, 0x0114) #define TSDB_CODE_REPEAT_INIT TAOS_DEF_ERROR_CODE(0, 0x0115) +#define TSDB_CODE_DUP_KEY TAOS_DEF_ERROR_CODE(0, 0x0116) #define TSDB_CODE_REF_NO_MEMORY TAOS_DEF_ERROR_CODE(0, 0x0140) #define TSDB_CODE_REF_FULL TAOS_DEF_ERROR_CODE(0, 0x0141) diff --git a/source/dnode/mgmt/implement/src/dmTransport.c b/source/dnode/mgmt/implement/src/dmTransport.c index 76cee15f01..10ab53eeee 100644 --- a/source/dnode/mgmt/implement/src/dmTransport.c +++ b/source/dnode/mgmt/implement/src/dmTransport.c @@ -72,6 +72,7 @@ static void dmProcessRpcMsg(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, SEpSet *pEpSe NodeMsgFp msgFp = NULL; uint16_t msgType = pRpc->msgType; bool needRelease = false; + bool isReq = msgType & 1U; if (pEpSet && pEpSet->numOfEps > 0 && msgType == TDMT_MND_STATUS_RSP) { dmSetMnodeEpSet(pWrapper->pDnode, pEpSet); @@ -85,13 +86,13 @@ static void dmProcessRpcMsg(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, SEpSet *pEpSe if (dmBuildMsg(pMsg, pRpc) != 0) goto _OVER; if (pWrapper->procType == DND_PROC_SINGLE) { - dTrace("msg:%p, is created, type:%s handle:%p user:%s", pMsg, TMSG_INFO(msgType), pRpc->handle, pMsg->user); + dTrace("msg:%p, created, type:%s handle:%p user:%s", pMsg, TMSG_INFO(msgType), pRpc->handle, pMsg->user); code = (*msgFp)(pWrapper, pMsg); } else if (pWrapper->procType == DND_PROC_PARENT) { - dTrace("msg:%p, is created and put into child queue, type:%s handle:%p user:%s", pMsg, TMSG_INFO(msgType), - pRpc->handle, pMsg->user); - code = taosProcPutToChildQ(pWrapper->procObj, pMsg, sizeof(SNodeMsg), pRpc->pCont, pRpc->contLen, pRpc->handle, - pRpc->refId, PROC_FUNC_REQ); + dTrace("msg:%p, created and put into child queue, type:%s handle:%p code:0x%04x user:%s contLen:%d", pMsg, + TMSG_INFO(msgType), pRpc->handle, pMsg->rpcMsg.code & 0XFFFF, pMsg->user, pRpc->contLen); + code = taosProcPutToChildQ(pWrapper->procObj, pMsg, sizeof(SNodeMsg), pRpc->pCont, pRpc->contLen, + (isReq && (pMsg->rpcMsg.code == 0)) ? pRpc->handle : NULL, pRpc->refId, PROC_FUNC_REQ); } else { dTrace("msg:%p, should not processed in child process, handle:%p user:%s", pMsg, pRpc->handle, pMsg->user); ASSERT(1); @@ -100,12 +101,13 @@ static void dmProcessRpcMsg(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, SEpSet *pEpSe _OVER: if (code == 0) { if (pWrapper->procType == DND_PROC_PARENT) { - dTrace("msg:%p, is freed in parent process", pMsg); + dTrace("msg:%p, freed in parent process", pMsg); taosFreeQitem(pMsg); rpcFreeCont(pRpc->pCont); } } else { - dError("msg:%p, type:%s failed to process since 0x%04x:%s", pMsg, TMSG_INFO(msgType), code & 0XFFFF, terrstr()); + dError("msg:%p, type:%s handle:%p failed to process since 0x%04x:%s", pMsg, TMSG_INFO(msgType), pRpc->handle, + code & 0XFFFF, terrstr()); if (msgType & 1U) { if (terrno != 0) code = terrno; if (code == TSDB_CODE_NODE_NOT_DEPLOYED || code == TSDB_CODE_NODE_OFFLINE) { @@ -359,29 +361,31 @@ static void dmConsumeChildQueue(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int16_t static void dmConsumeParentQueue(SMgmtWrapper *pWrapper, SRpcMsg *pMsg, int16_t msgLen, void *pCont, int32_t contLen, EProcFuncType ftype) { + int32_t code = pMsg->code & 0xFFFF; pMsg->pCont = pCont; - dTrace("msg:%p, get from parent queue, ftype:%d handle:%p code:0x%04x mtype:%d, app:%p", pMsg, ftype, pMsg->handle, - pMsg->code & 0xFFFF, pMsg->msgType, pMsg->ahandle); - switch (ftype) { - case PROC_FUNC_REGIST: - rpcRegisterBrokenLinkArg(pMsg); - break; - case PROC_FUNC_RELEASE: - taosProcRemoveHandle(pWrapper->procObj, pMsg->handle); - rpcReleaseHandle(pMsg->handle, (int8_t)pMsg->code); - rpcFreeCont(pCont); - break; - case PROC_FUNC_REQ: - dmSendRpcReq(pWrapper->pDnode, (SEpSet *)((char *)pMsg + sizeof(SRpcMsg)), pMsg); - break; - case PROC_FUNC_RSP: - pMsg->refId = taosProcRemoveHandle(pWrapper->procObj, pMsg->handle); - dmSendRpcRsp(pWrapper->pDnode, pMsg); - break; - default: - break; + if (ftype == PROC_FUNC_REQ) { + dTrace("msg:%p, get from parent queue, send req:%s handle:%p code:0x%04x, app:%p", pMsg, TMSG_INFO(pMsg->msgType), + pMsg->handle, code, pMsg->ahandle); + dmSendRpcReq(pWrapper->pDnode, (SEpSet *)((char *)pMsg + sizeof(SRpcMsg)), pMsg); + } else if (ftype == PROC_FUNC_RSP) { + dTrace("msg:%p, get from parent queue, rsp handle:%p code:0x%04x, app:%p", pMsg, pMsg->handle, code, pMsg->ahandle); + pMsg->refId = taosProcRemoveHandle(pWrapper->procObj, pMsg->handle); + dmSendRpcRsp(pWrapper->pDnode, pMsg); + } else if (ftype == PROC_FUNC_REGIST) { + dTrace("msg:%p, get from parent queue, regist handle:%p code:0x%04x, app:%p", pMsg, pMsg->handle, code, + pMsg->ahandle); + rpcRegisterBrokenLinkArg(pMsg); + } else if (ftype == PROC_FUNC_RELEASE) { + dTrace("msg:%p, get from parent queue, release handle:%p code:0x%04x, app:%p", pMsg, pMsg->handle, code, + pMsg->ahandle); + taosProcRemoveHandle(pWrapper->procObj, pMsg->handle); + rpcReleaseHandle(pMsg->handle, (int8_t)pMsg->code); + rpcFreeCont(pCont); + } else { + dError("msg:%p, invalid ftype:%d while get from parent queue, handle:%p", pMsg, ftype, pMsg->handle); } + taosMemoryFree(pMsg); } diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c index 7285503a73..818fd00430 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c @@ -136,7 +136,7 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO // sync integration response for (int i = 0; i < taosArrayGetSize(pArray); i++) { SNodeMsg *pMsg; - SRpcMsg * pRpc; + SRpcMsg *pRpc; pMsg = *(SNodeMsg **)taosArrayGet(pArray, i); pRpc = &pMsg->rpcMsg; @@ -175,7 +175,7 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) { SVnodeObj *pVnode = pInfo->ahandle; - SNodeMsg * pMsg = NULL; + SNodeMsg *pMsg = NULL; SRpcMsg rsp; for (int32_t i = 0; i < numOfMsgs; ++i) { @@ -218,7 +218,7 @@ static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) { SVnodeObj *pVnode = pInfo->ahandle; - SNodeMsg * pMsg = NULL; + SNodeMsg *pMsg = NULL; for (int32_t i = 0; i < numOfMsgs; ++i) { taosGetQitem(qall, (void **)&pMsg); @@ -231,7 +231,7 @@ static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOf static void vmProcessMergeQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) { SVnodeObj *pVnode = pInfo->ahandle; - SNodeMsg * pMsg = NULL; + SNodeMsg *pMsg = NULL; for (int32_t i = 0; i < numOfMsgs; ++i) { taosGetQitem(qall, (void **)&pMsg); @@ -248,7 +248,7 @@ static void vmProcessMergeQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO } static int32_t vmPutNodeMsgToQueue(SVnodesMgmt *pMgmt, SNodeMsg *pMsg, EQueueType qtype) { - SRpcMsg * pRpc = &pMsg->rpcMsg; + SRpcMsg *pRpc = &pMsg->rpcMsg; SMsgHead *pHead = pRpc->pCont; pHead->contLen = ntohl(pHead->contLen); pHead->vgId = ntohl(pHead->vgId); @@ -262,23 +262,23 @@ static int32_t vmPutNodeMsgToQueue(SVnodesMgmt *pMgmt, SNodeMsg *pMsg, EQueueTyp int32_t code = 0; switch (qtype) { case QUERY_QUEUE: - dTrace("msg:%p, will be written into vnode-query queue", pMsg); + dTrace("msg:%p, type:%s will be written into vnode-query queue", pMsg, TMSG_INFO(pRpc->msgType)); taosWriteQitem(pVnode->pQueryQ, pMsg); break; case FETCH_QUEUE: - dTrace("msg:%p, will be written into vnode-fetch queue", pMsg); + dTrace("msg:%p, type:%s will be written into vnode-fetch queue", pMsg, TMSG_INFO(pRpc->msgType)); taosWriteQitem(pVnode->pFetchQ, pMsg); break; case WRITE_QUEUE: - dTrace("msg:%p, will be written into vnode-write queue", pMsg); + dTrace("msg:%p, type:%s will be written into vnode-write queue", pMsg, TMSG_INFO(pRpc->msgType)); taosWriteQitem(pVnode->pWriteQ, pMsg); break; case SYNC_QUEUE: - dTrace("msg:%p, will be written into vnode-sync queue", pMsg); + dTrace("msg:%p, type:%s will be written into vnode-sync queue", pMsg, TMSG_INFO(pRpc->msgType)); taosWriteQitem(pVnode->pSyncQ, pMsg); break; case MERGE_QUEUE: - dTrace("msg:%p, will be written into vnode-merge queue", pMsg); + dTrace("msg:%p, type:%s will be written into vnode-merge queue", pMsg, TMSG_INFO(pRpc->msgType)); taosWriteQitem(pVnode->pMergeQ, pMsg); break; default: @@ -317,7 +317,7 @@ int32_t vmProcessMergeMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { } int32_t vmProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { - SVnodesMgmt * pMgmt = pWrapper->pMgmt; + SVnodesMgmt *pMgmt = pWrapper->pMgmt; SSingleWorker *pWorker = &pMgmt->mgmtWorker; dTrace("msg:%p, will be written to vnode-mgmt queue, worker:%s", pMsg, pWorker->name); taosWriteQitem(pWorker->queue, pMsg); @@ -325,7 +325,7 @@ int32_t vmProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { } int32_t vmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { - SVnodesMgmt * pMgmt = pWrapper->pMgmt; + SVnodesMgmt *pMgmt = pWrapper->pMgmt; SSingleWorker *pWorker = &pMgmt->monitorWorker; dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name); @@ -335,7 +335,7 @@ int32_t vmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) { static int32_t vmPutRpcMsgToQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, EQueueType qtype) { SVnodesMgmt *pMgmt = pWrapper->pMgmt; - SMsgHead * pHead = pRpc->pCont; + SMsgHead *pHead = pRpc->pCont; SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId); if (pVnode == NULL) return -1; diff --git a/source/dnode/mnode/impl/src/mndMnode.c b/source/dnode/mnode/impl/src/mndMnode.c index 465bf4d5a1..b7d7358110 100644 --- a/source/dnode/mnode/impl/src/mndMnode.c +++ b/source/dnode/mnode/impl/src/mndMnode.c @@ -625,7 +625,7 @@ static int32_t mndRetrieveMnodes(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock *p colDataAppend(pColInfo, numOfRows, b1, false); const char *roles = syncStr(pObj->role); - char *b2 = taosMemoryCalloc(1, strlen(roles) + VARSTR_HEADER_SIZE); + char *b2 = taosMemoryCalloc(1, 12 + VARSTR_HEADER_SIZE); STR_WITH_MAXSIZE_TO_VARSTR(b2, roles, pShow->pMeta->pSchemas[cols].bytes); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); diff --git a/source/dnode/mnode/impl/src/mndStb.c b/source/dnode/mnode/impl/src/mndStb.c index ec8dfa4d47..1349439dd2 100644 --- a/source/dnode/mnode/impl/src/mndStb.c +++ b/source/dnode/mnode/impl/src/mndStb.c @@ -679,8 +679,8 @@ static int32_t mndCreateStb(SMnode *pMnode, SNodeMsg *pReq, SMCreateStbReq *pCre memcpy(stbObj.pAst2, pCreate->pAst2, stbObj.ast2Len); } - stbObj.pColumns = taosMemoryMalloc(stbObj.numOfColumns * sizeof(SSchema)); - stbObj.pTags = taosMemoryMalloc(stbObj.numOfTags * sizeof(SSchema)); + stbObj.pColumns = taosMemoryCalloc(1, stbObj.numOfColumns * sizeof(SSchema)); + stbObj.pTags = taosMemoryCalloc(1, stbObj.numOfTags * sizeof(SSchema)); if (stbObj.pColumns == NULL || stbObj.pTags == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; return -1; @@ -1111,7 +1111,7 @@ static int32_t mndSetAlterStbRedoLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pD SSdbRaw *pRedoRaw = mndStbActionEncode(pStb); if (pRedoRaw == NULL) return -1; if (mndTransAppendRedolog(pTrans, pRedoRaw) != 0) return -1; - if (sdbSetRawStatus(pRedoRaw, SDB_STATUS_UPDATING) != 0) return -1; + if (sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY) != 0) return -1; return 0; } diff --git a/source/dnode/mnode/impl/test/sdb/sdbTest.cpp b/source/dnode/mnode/impl/test/sdb/sdbTest.cpp index 31d4733fe1..b93adf9930 100644 --- a/source/dnode/mnode/impl/test/sdb/sdbTest.cpp +++ b/source/dnode/mnode/impl/test/sdb/sdbTest.cpp @@ -15,8 +15,31 @@ class MndTestSdb : public ::testing::Test { protected: - static void SetUpTestSuite() {} - static void TearDownTestSuite() {} + static void SetUpTestSuite() { + dDebugFlag = 143; + vDebugFlag = 0; + mDebugFlag = 143; + cDebugFlag = 0; + jniDebugFlag = 0; + tmrDebugFlag = 135; + uDebugFlag = 135; + rpcDebugFlag = 143; + qDebugFlag = 0; + wDebugFlag = 0; + sDebugFlag = 0; + tsdbDebugFlag = 0; + tsLogEmbedded = 1; + tsAsyncLog = 0; + + const char *path = "/tmp/td"; + taosRemoveDir(path); + taosMkDir(path); + tstrncpy(tsLogDir, path, PATH_MAX); + if (taosInitLog("taosdlog", 1) != 0) { + printf("failed to init log file\n"); + } + } + static void TearDownTestSuite() { taosCloseLog(); } public: void SetUp() override {} @@ -26,6 +49,8 @@ class MndTestSdb : public ::testing::Test { typedef struct SMnode { int32_t v100; int32_t v200; + int32_t insertTimes; + int32_t deleteTimes; SSdb *pSdb; } SMnode; @@ -80,6 +105,48 @@ SSdbRaw *strEncode(SStrObj *pObj) { return pRaw; } +SSdbRaw *i32Encode(SI32Obj *pObj) { + int32_t dataPos = 0; + SSdbRaw *pRaw = sdbAllocRaw(SDB_VGROUP, 2, sizeof(SI32Obj)); + + sdbSetRawInt32(pRaw, dataPos, pObj->key); + dataPos += sizeof(pObj->key); + sdbSetRawInt8(pRaw, dataPos, pObj->v8); + dataPos += sizeof(pObj->v8); + sdbSetRawInt16(pRaw, dataPos, pObj->v16); + dataPos += sizeof(pObj->v16); + sdbSetRawInt32(pRaw, dataPos, pObj->v32); + dataPos += sizeof(pObj->v32); + sdbSetRawInt64(pRaw, dataPos, pObj->v64); + dataPos += sizeof(pObj->v64); + sdbSetRawBinary(pRaw, dataPos, pObj->vstr, sizeof(pObj->vstr)); + dataPos += sizeof(pObj->vstr); + sdbSetRawDataLen(pRaw, dataPos); + + return pRaw; +} + +SSdbRaw *i64Encode(SI64Obj *pObj) { + int32_t dataPos = 0; + SSdbRaw *pRaw = sdbAllocRaw(SDB_CONSUMER, 3, sizeof(SI64Obj)); + + sdbSetRawInt64(pRaw, dataPos, pObj->key); + dataPos += sizeof(pObj->key); + sdbSetRawInt8(pRaw, dataPos, pObj->v8); + dataPos += sizeof(pObj->v8); + sdbSetRawInt16(pRaw, dataPos, pObj->v16); + dataPos += sizeof(pObj->v16); + sdbSetRawInt32(pRaw, dataPos, pObj->v32); + dataPos += sizeof(pObj->v32); + sdbSetRawInt64(pRaw, dataPos, pObj->v64); + dataPos += sizeof(pObj->v64); + sdbSetRawBinary(pRaw, dataPos, pObj->vstr, sizeof(pObj->vstr)); + dataPos += sizeof(pObj->vstr); + sdbSetRawDataLen(pRaw, dataPos); + + return pRaw; +} + SSdbRow *strDecode(SSdbRaw *pRaw) { int8_t sver = 0; if (sdbGetRawSoftVer(pRaw, &sver) != 0) return NULL; @@ -108,9 +175,103 @@ SSdbRow *strDecode(SSdbRaw *pRaw) { return pRow; } -int32_t strInsert(SSdb *pSdb, SStrObj *pObj) { return 0; } +SSdbRow *i32Decode(SSdbRaw *pRaw) { + int8_t sver = 0; + if (sdbGetRawSoftVer(pRaw, &sver) != 0) return NULL; + if (sver != 2) return NULL; -int32_t strDelete(SSdb *pSdb, SStrObj *pObj, bool callFunc) { return 0; } + SSdbRow *pRow = sdbAllocRow(sizeof(SI32Obj)); + if (pRow == NULL) return NULL; + + SI32Obj *pObj = (SI32Obj *)sdbGetRowObj(pRow); + if (pObj == NULL) return NULL; + + int32_t dataPos = 0; + sdbGetRawInt32(pRaw, dataPos, &pObj->key); + dataPos += sizeof(pObj->key); + sdbGetRawInt8(pRaw, dataPos, &pObj->v8); + dataPos += sizeof(pObj->v8); + sdbGetRawInt16(pRaw, dataPos, &pObj->v16); + dataPos += sizeof(pObj->v16); + sdbGetRawInt32(pRaw, dataPos, &pObj->v32); + dataPos += sizeof(pObj->v32); + sdbGetRawInt64(pRaw, dataPos, &pObj->v64); + dataPos += sizeof(pObj->v64); + sdbGetRawBinary(pRaw, dataPos, pObj->vstr, sizeof(pObj->vstr)); + dataPos += sizeof(pObj->vstr); + + return pRow; +} + +SSdbRow *i64Decode(SSdbRaw *pRaw) { + int8_t sver = 0; + if (sdbGetRawSoftVer(pRaw, &sver) != 0) return NULL; + if (sver != 3) return NULL; + + SSdbRow *pRow = sdbAllocRow(sizeof(SI64Obj)); + if (pRow == NULL) return NULL; + + SI64Obj *pObj = (SI64Obj *)sdbGetRowObj(pRow); + if (pObj == NULL) return NULL; + + int32_t dataPos = 0; + sdbGetRawInt64(pRaw, dataPos, &pObj->key); + dataPos += sizeof(pObj->key); + sdbGetRawInt8(pRaw, dataPos, &pObj->v8); + dataPos += sizeof(pObj->v8); + sdbGetRawInt16(pRaw, dataPos, &pObj->v16); + dataPos += sizeof(pObj->v16); + sdbGetRawInt32(pRaw, dataPos, &pObj->v32); + dataPos += sizeof(pObj->v32); + sdbGetRawInt64(pRaw, dataPos, &pObj->v64); + dataPos += sizeof(pObj->v64); + sdbGetRawBinary(pRaw, dataPos, pObj->vstr, sizeof(pObj->vstr)); + dataPos += sizeof(pObj->vstr); + + return pRow; +} + +int32_t strInsert(SSdb *pSdb, SStrObj *pObj) { + SMnode *pMnode = pSdb->pMnode; + pMnode->insertTimes++; + return 0; +} + +int32_t i32Insert(SSdb *pSdb, SI32Obj *pObj) { + SMnode *pMnode = pSdb->pMnode; + pMnode->insertTimes++; + return 0; +} + +int32_t i64Insert(SSdb *pSdb, SI64Obj *pObj) { + SMnode *pMnode = pSdb->pMnode; + pMnode->insertTimes++; + return 0; +} + +int32_t strDelete(SSdb *pSdb, SStrObj *pObj, bool callFunc) { + if (callFunc) { + SMnode *pMnode = pSdb->pMnode; + pMnode->deleteTimes++; + } + return 0; +} + +int32_t i32Delete(SSdb *pSdb, SI32Obj *pObj, bool callFunc) { + if (callFunc) { + SMnode *pMnode = pSdb->pMnode; + pMnode->deleteTimes++; + } + return 0; +} + +int32_t i64Delete(SSdb *pSdb, SI64Obj *pObj, bool callFunc) { + if (callFunc) { + SMnode *pMnode = pSdb->pMnode; + pMnode->deleteTimes++; + } + return 0; +} int32_t strUpdate(SSdb *pSdb, SStrObj *pOld, SStrObj *pNew) { pOld->v8 = pNew->v8; @@ -121,6 +282,24 @@ int32_t strUpdate(SSdb *pSdb, SStrObj *pOld, SStrObj *pNew) { return 0; } +int32_t i32Update(SSdb *pSdb, SI32Obj *pOld, SI32Obj *pNew) { + pOld->v8 = pNew->v8; + pOld->v16 = pNew->v16; + pOld->v32 = pNew->v32; + pOld->v64 = pNew->v64; + strcpy(pOld->vstr, pNew->vstr); + return 0; +} + +int32_t i64Update(SSdb *pSdb, SI64Obj *pOld, SI64Obj *pNew) { + pOld->v8 = pNew->v8; + pOld->v16 = pNew->v16; + pOld->v32 = pNew->v32; + pOld->v64 = pNew->v64; + strcpy(pOld->vstr, pNew->vstr); + return 0; +} + void strSetDefault(SStrObj *pObj, int32_t index) { memset(pObj, 0, sizeof(SStrObj)); snprintf(pObj->key, sizeof(pObj->key), "k%d", index * 1000); @@ -131,6 +310,26 @@ void strSetDefault(SStrObj *pObj, int32_t index) { snprintf(pObj->vstr, sizeof(pObj->vstr), "v%d", index * 1000); } +void i32SetDefault(SI32Obj *pObj, int32_t index) { + memset(pObj, 0, sizeof(SI32Obj)); + pObj->key = index; + pObj->v8 = index; + pObj->v16 = index; + pObj->v32 = index * 1000; + pObj->v64 = index * 1000; + snprintf(pObj->vstr, sizeof(pObj->vstr), "v%d", index * 1000); +} + +void i64SetDefault(SI64Obj *pObj, int32_t index) { + memset(pObj, 0, sizeof(SI64Obj)); + pObj->key = index; + pObj->v8 = index; + pObj->v16 = index; + pObj->v32 = index * 1000; + pObj->v64 = index * 1000; + snprintf(pObj->vstr, sizeof(pObj->vstr), "v%d", index * 1000); +} + int32_t strDefault(SMnode *pMnode) { SStrObj strObj; SSdbRaw *pRaw = NULL; @@ -144,6 +343,8 @@ int32_t strDefault(SMnode *pMnode) { pRaw = strEncode(&strObj); sdbSetRawStatus(pRaw, SDB_STATUS_READY); if (sdbWriteWithoutFree(pMnode->pSdb, pRaw) != 0) return -1; + + EXPECT_EQ(sdbGetRawTotalSize(pRaw), 79); sdbFreeRaw(pRaw); return 0; @@ -151,28 +352,32 @@ int32_t strDefault(SMnode *pMnode) { bool sdbTraverseSucc1(SMnode *pMnode, SStrObj *pObj, int32_t *p1, int32_t *p2, int32_t *p3) { if (pObj->v8 == 1) { - *p1 = *p2 + *p3 + pObj->v8; + *p1 += *p2 + *p3 + pObj->v8; } return true; } bool sdbTraverseSucc2(SMnode *pMnode, SStrObj *pObj, int32_t *p1, int32_t *p2, int32_t *p3) { - *p1 = *p2 + *p3 + pObj->v8; + *p1 += *p2 + *p3 + pObj->v8; return true; } bool sdbTraverseFail(SMnode *pMnode, SStrObj *pObj, int32_t *p1, int32_t *p2, int32_t *p3) { - *p1 = *p2 + *p3; + *p1 += *p2 + *p3; return false; } -TEST_F(MndTestSdb, 01_Write) { - void *pIter; - int32_t num; - SStrObj *pObj; - SMnode mnode; - SSdb *pSdb; +TEST_F(MndTestSdb, 01_Write_Str) { + void *pIter = NULL; + int32_t num = 0; + SStrObj *pObj = NULL; + SMnode mnode = {0}; + SSdb *pSdb = NULL; SSdbOpt opt = {0}; + SStrObj strObj = {0}; + SI32Obj i32Obj = {0}; + SI64Obj i64Obj = {0}; + SSdbRaw *pRaw = NULL; int32_t p1 = 0; int32_t p2 = 111; int32_t p3 = 222; @@ -183,42 +388,64 @@ TEST_F(MndTestSdb, 01_Write) { opt.path = "/tmp/mnode_test_sdb"; taosRemoveDir(opt.path); - SSdbTable strTable = { - SDB_USER, - SDB_KEY_BINARY, - (SdbDeployFp)strDefault, - (SdbEncodeFp)strEncode, - (SdbDecodeFp)strDecode, - (SdbInsertFp)strInsert, - (SdbUpdateFp)strUpdate, - (SdbDeleteFp)strDelete, - }; + SSdbTable strTable1; + memset(&strTable1, 0, sizeof(SSdbTable)); + strTable1.sdbType = SDB_USER; + strTable1.keyType = SDB_KEY_BINARY; + strTable1.deployFp = (SdbDeployFp)strDefault; + strTable1.encodeFp = (SdbEncodeFp)strEncode; + strTable1.decodeFp = (SdbDecodeFp)strDecode; + strTable1.insertFp = (SdbInsertFp)strInsert; + strTable1.updateFp = (SdbUpdateFp)strUpdate; + strTable1.deleteFp = (SdbDeleteFp)strDelete; + + SSdbTable strTable2; + memset(&strTable2, 0, sizeof(SSdbTable)); + strTable2.sdbType = SDB_VGROUP; + strTable2.keyType = SDB_KEY_INT32; + strTable2.encodeFp = (SdbEncodeFp)i32Encode; + strTable2.decodeFp = (SdbDecodeFp)i32Decode; + strTable2.insertFp = (SdbInsertFp)i32Insert; + strTable2.updateFp = (SdbUpdateFp)i32Update; + strTable2.deleteFp = (SdbDeleteFp)i32Delete; + + SSdbTable strTable3; + memset(&strTable3, 0, sizeof(SSdbTable)); + strTable3.sdbType = SDB_CONSUMER; + strTable3.keyType = SDB_KEY_INT64; + strTable3.encodeFp = (SdbEncodeFp)i64Encode; + strTable3.decodeFp = (SdbDecodeFp)i64Decode; + strTable3.insertFp = (SdbInsertFp)i64Insert; + strTable3.updateFp = (SdbUpdateFp)i64Update; + strTable3.deleteFp = (SdbDeleteFp)i64Delete; pSdb = sdbInit(&opt); mnode.pSdb = pSdb; ASSERT_NE(pSdb, nullptr); - ASSERT_EQ(sdbSetTable(pSdb, strTable), 0); + ASSERT_EQ(sdbSetTable(pSdb, strTable1), 0); + ASSERT_EQ(sdbSetTable(pSdb, strTable2), 0); + ASSERT_EQ(sdbSetTable(pSdb, strTable3), 0); ASSERT_EQ(sdbDeploy(pSdb), 0); -#if 0 + pObj = (SStrObj *)sdbAcquire(pSdb, SDB_USER, "k1000"); ASSERT_NE(pObj, nullptr); EXPECT_STREQ(pObj->key, "k1000"); EXPECT_STREQ(pObj->vstr, "v1000"); - EXPECT_EQ(pObj->v8, 1); - EXPECT_EQ(pObj->v16, 1); - EXPECT_EQ(pObj->v32, 1000); - EXPECT_EQ(pObj->v64, 1000); + ASSERT_EQ(pObj->v8, 1); + ASSERT_EQ(pObj->v16, 1); + ASSERT_EQ(pObj->v32, 1000); + ASSERT_EQ(pObj->v64, 1000); sdbRelease(pSdb, pObj); pObj = (SStrObj *)sdbAcquire(pSdb, SDB_USER, "k2000"); ASSERT_NE(pObj, nullptr); EXPECT_STREQ(pObj->key, "k2000"); EXPECT_STREQ(pObj->vstr, "v2000"); - EXPECT_EQ(pObj->v8, 2); - EXPECT_EQ(pObj->v16, 2); - EXPECT_EQ(pObj->v32, 2000); - EXPECT_EQ(pObj->v64, 2000); + ASSERT_EQ(pObj->v8, 2); + ASSERT_EQ(pObj->v16, 2); + ASSERT_EQ(pObj->v32, 2000); + ASSERT_EQ(pObj->v64, 2000); sdbRelease(pSdb, pObj); pObj = (SStrObj *)sdbAcquire(pSdb, SDB_USER, "k200"); @@ -233,7 +460,7 @@ TEST_F(MndTestSdb, 01_Write) { num++; sdbRelease(pSdb, pObj); } while (1); - EXPECT_EQ(num, 2); + ASSERT_EQ(num, 2); do { pIter = sdbFetch(pSdb, SDB_USER, pIter, (void **)&pObj); @@ -248,51 +475,254 @@ TEST_F(MndTestSdb, 01_Write) { p1 = 0; p2 = 111; p3 = 222; - sdbTraverse(pSdb, SDB_USER, (sdbTraverseFp)sdbTraverseSucc2, &p1, &p2, &p3); - EXPECT_EQ(p1, 334); + sdbTraverse(pSdb, SDB_USER, (sdbTraverseFp)sdbTraverseSucc1, &p1, &p2, &p3); + ASSERT_EQ(p1, 334); p1 = 0; p2 = 111; p3 = 222; sdbTraverse(pSdb, SDB_USER, (sdbTraverseFp)sdbTraverseSucc2, &p1, &p2, &p3); - EXPECT_EQ(p1, 669); + ASSERT_EQ(p1, 669); p1 = 0; p2 = 111; p3 = 222; sdbTraverse(pSdb, SDB_USER, (sdbTraverseFp)sdbTraverseFail, &p1, &p2, &p3); - EXPECT_EQ(p1, 333); + ASSERT_EQ(p1, 333); - EXPECT_EQ(sdbGetSize(pSdb, SDB_USER), 2); - EXPECT_EQ(sdbGetMaxId(pSdb, SDB_USER), -1); - EXPECT_EQ(sdbGetTableVer(pSdb, SDB_USER), 2); - EXPECT_EQ(sdbUpdateVer(pSdb, 0), 2); - EXPECT_EQ(sdbUpdateVer(pSdb, 1), 3); - EXPECT_EQ(sdbUpdateVer(pSdb, -1), 2); + ASSERT_EQ(sdbGetSize(pSdb, SDB_USER), 2); + ASSERT_EQ(sdbGetMaxId(pSdb, SDB_USER), -1); + ASSERT_EQ(sdbGetTableVer(pSdb, SDB_USER), 2 ); + ASSERT_EQ(sdbUpdateVer(pSdb, 0), -1); + ASSERT_EQ(sdbUpdateVer(pSdb, 1), 0); + ASSERT_EQ(sdbUpdateVer(pSdb, -1), -1); + ASSERT_EQ(mnode.insertTimes, 2); + ASSERT_EQ(mnode.deleteTimes, 0); - // insert, call func + { + // insert, call func + strSetDefault(&strObj, 3); + pRaw = strEncode(&strObj); + sdbSetRawStatus(pRaw, SDB_STATUS_READY); + ASSERT_EQ(sdbWrite(pSdb, pRaw), 0); + pObj = (SStrObj *)sdbAcquire(pSdb, SDB_USER, "k3000"); + ASSERT_NE(pObj, nullptr); + EXPECT_STREQ(pObj->key, "k3000"); + EXPECT_STREQ(pObj->vstr, "v3000"); + ASSERT_EQ(pObj->v8, 3); + ASSERT_EQ(pObj->v16, 3); + ASSERT_EQ(pObj->v32, 3000); + ASSERT_EQ(pObj->v64, 3000); + sdbRelease(pSdb, pObj); - // update, call func + ASSERT_EQ(sdbGetSize(pSdb, SDB_USER), 3); + ASSERT_EQ(sdbGetTableVer(pSdb, SDB_USER), 3); + ASSERT_EQ(sdbGetMaxId(pSdb, SDB_USER), -1); - // delete, call func 2 + // update, call func + strSetDefault(&strObj, 3); + strObj.v8 = 4; + pRaw = strEncode(&strObj); + sdbSetRawStatus(pRaw, SDB_STATUS_READY); + ASSERT_EQ(sdbWrite(pSdb, pRaw), 0); + pObj = (SStrObj *)sdbAcquire(pSdb, SDB_USER, "k3000"); + ASSERT_NE(pObj, nullptr); + EXPECT_STREQ(pObj->key, "k3000"); + EXPECT_STREQ(pObj->vstr, "v3000"); + ASSERT_EQ(pObj->v8, 4); + ASSERT_EQ(pObj->v16, 3); + ASSERT_EQ(pObj->v32, 3000); + ASSERT_EQ(pObj->v64, 3000); + sdbRelease(pSdb, pObj); + + ASSERT_EQ(sdbGetSize(pSdb, SDB_USER), 3); + ASSERT_EQ(sdbGetTableVer(pSdb, SDB_USER), 4); + ASSERT_EQ(sdbUpdateVer(pSdb, 0), -1); + ASSERT_EQ(sdbUpdateVer(pSdb, 1), 0); + ASSERT_EQ(sdbUpdateVer(pSdb, -1), -1); + ASSERT_EQ(mnode.insertTimes, 3); + ASSERT_EQ(mnode.deleteTimes, 0); + + // delete, call func 2 + strSetDefault(&strObj, 3); + strObj.v16 = 4; + pRaw = strEncode(&strObj); + sdbSetRawStatus(pRaw, SDB_STATUS_DROPPED); + ASSERT_EQ(sdbWrite(pSdb, pRaw), 0); + pObj = (SStrObj *)sdbAcquire(pSdb, SDB_USER, "k3000"); + ASSERT_EQ(pObj, nullptr); + ASSERT_EQ(sdbGetSize(pSdb, SDB_USER), 2); + ASSERT_EQ(sdbGetTableVer(pSdb, SDB_USER), 5); + ASSERT_EQ(mnode.insertTimes, 3); + ASSERT_EQ(mnode.deleteTimes, 1); + } + + { + int32_t key = 4; + i32SetDefault(&i32Obj, key); + pRaw = i32Encode(&i32Obj); + sdbSetRawStatus(pRaw, SDB_STATUS_READY); + ASSERT_EQ(sdbWrite(pSdb, pRaw), 0); + SI32Obj *pI32Obj = (SI32Obj *)sdbAcquire(pSdb, SDB_VGROUP, &key); + ASSERT_NE(pI32Obj, nullptr); + ASSERT_EQ(pI32Obj->key, key); + ASSERT_EQ(pI32Obj->v8, 4); + ASSERT_EQ(pI32Obj->v16, 4); + ASSERT_EQ(pI32Obj->v32, 4000); + ASSERT_EQ(pI32Obj->v64, 4000); + sdbRelease(pSdb, pI32Obj); + + ASSERT_EQ(sdbGetSize(pSdb, SDB_VGROUP), 1); + ASSERT_EQ(sdbGetTableVer(pSdb, SDB_VGROUP), 1); + ASSERT_EQ(sdbGetMaxId(pSdb, SDB_VGROUP), 5); + + i32SetDefault(&i32Obj, key); + i32Obj.v8 = 5; + pRaw = i32Encode(&i32Obj); + sdbSetRawStatus(pRaw, SDB_STATUS_READY); + ASSERT_EQ(sdbWrite(pSdb, pRaw), 0); + pI32Obj = (SI32Obj *)sdbAcquire(pSdb, SDB_VGROUP, &key); + ASSERT_NE(pI32Obj, nullptr); + ASSERT_EQ(pI32Obj->key, key); + ASSERT_EQ(pI32Obj->v8, 5); + ASSERT_EQ(pI32Obj->v16, 4); + ASSERT_EQ(pI32Obj->v32, 4000); + ASSERT_EQ(pI32Obj->v64, 4000); + sdbRelease(pSdb, pI32Obj); + + ASSERT_EQ(sdbGetSize(pSdb, SDB_VGROUP), 1); + ASSERT_EQ(sdbGetTableVer(pSdb, SDB_VGROUP), 2); + ASSERT_EQ(mnode.insertTimes, 4); + ASSERT_EQ(mnode.deleteTimes, 1); + + // delete, call func 2 + key = 4; + i32SetDefault(&i32Obj, key); + pRaw = i32Encode(&i32Obj); + sdbSetRawStatus(pRaw, SDB_STATUS_DROPPED); + ASSERT_EQ(sdbWrite(pSdb, pRaw), 0); + pI32Obj = (SI32Obj *)sdbAcquire(pSdb, SDB_VGROUP, &key); + ASSERT_EQ(pI32Obj, nullptr); + ASSERT_EQ(sdbGetSize(pSdb, SDB_VGROUP), 0); + ASSERT_EQ(sdbGetTableVer(pSdb, SDB_VGROUP), 3); + ASSERT_EQ(sdbGetMaxId(pSdb, SDB_VGROUP), 5); + ASSERT_EQ(mnode.insertTimes, 4); + ASSERT_EQ(mnode.deleteTimes, 2); + + key = 6; + i32SetDefault(&i32Obj, key); + pRaw = i32Encode(&i32Obj); + sdbSetRawStatus(pRaw, SDB_STATUS_READY); + ASSERT_EQ(sdbWrite(pSdb, pRaw), 0); + pI32Obj = (SI32Obj *)sdbAcquire(pSdb, SDB_VGROUP, &key); + ASSERT_NE(pI32Obj, nullptr); + ASSERT_EQ(pI32Obj->key, key); + ASSERT_EQ(pI32Obj->v8, 6); + ASSERT_EQ(pI32Obj->v16, 6); + ASSERT_EQ(pI32Obj->v32, 6000); + ASSERT_EQ(pI32Obj->v64, 6000); + sdbRelease(pSdb, pI32Obj); + + ASSERT_EQ(sdbGetSize(pSdb, SDB_VGROUP), 1); + ASSERT_EQ(sdbGetTableVer(pSdb, SDB_VGROUP), 4); + ASSERT_EQ(sdbGetMaxId(pSdb, SDB_VGROUP), 7); + ASSERT_EQ(mnode.insertTimes, 5); + ASSERT_EQ(mnode.deleteTimes, 2); + } + + { + int64_t key = 4; + i64SetDefault(&i64Obj, key); + pRaw = i64Encode(&i64Obj); + sdbSetRawStatus(pRaw, SDB_STATUS_READY); + ASSERT_EQ(sdbWrite(pSdb, pRaw), 0); + SI64Obj *pI64Obj = (SI64Obj *)sdbAcquire(pSdb, SDB_CONSUMER, &key); + ASSERT_NE(pI64Obj, nullptr); + ASSERT_EQ(pI64Obj->key, key); + ASSERT_EQ(pI64Obj->v8, 4); + ASSERT_EQ(pI64Obj->v16, 4); + ASSERT_EQ(pI64Obj->v32, 4000); + ASSERT_EQ(pI64Obj->v64, 4000); + sdbRelease(pSdb, pI64Obj); + + ASSERT_EQ(sdbGetSize(pSdb, SDB_CONSUMER), 1); + ASSERT_EQ(sdbGetTableVer(pSdb, SDB_CONSUMER), 1); + ASSERT_EQ(sdbGetMaxId(pSdb, SDB_CONSUMER), -1); + + i64SetDefault(&i64Obj, key); + i64Obj.v8 = 5; + pRaw = i64Encode(&i64Obj); + sdbSetRawStatus(pRaw, SDB_STATUS_READY); + ASSERT_EQ(sdbWrite(pSdb, pRaw), 0); + pI64Obj = (SI64Obj *)sdbAcquire(pSdb, SDB_CONSUMER, &key); + ASSERT_NE(pI64Obj, nullptr); + ASSERT_EQ(pI64Obj->key, key); + ASSERT_EQ(pI64Obj->v8, 5); + ASSERT_EQ(pI64Obj->v16, 4); + ASSERT_EQ(pI64Obj->v32, 4000); + ASSERT_EQ(pI64Obj->v64, 4000); + sdbRelease(pSdb, pI64Obj); + + ASSERT_EQ(sdbGetSize(pSdb, SDB_CONSUMER), 1); + ASSERT_EQ(sdbGetTableVer(pSdb, SDB_CONSUMER), 2); + ASSERT_EQ(mnode.insertTimes, 6); + ASSERT_EQ(mnode.deleteTimes, 2); + + // delete, call func 2 + key = 4; + i64SetDefault(&i64Obj, key); + pRaw = i64Encode(&i64Obj); + sdbSetRawStatus(pRaw, SDB_STATUS_DROPPED); + ASSERT_EQ(sdbWrite(pSdb, pRaw), 0); + pObj = (SStrObj *)sdbAcquire(pSdb, SDB_CONSUMER, &key); + ASSERT_EQ(pObj, nullptr); + ASSERT_EQ(sdbGetSize(pSdb, SDB_CONSUMER), 0); + ASSERT_EQ(sdbGetTableVer(pSdb, SDB_CONSUMER), 3); + ASSERT_EQ(sdbGetMaxId(pSdb, SDB_CONSUMER), -1); + ASSERT_EQ(mnode.insertTimes, 6); + ASSERT_EQ(mnode.deleteTimes, 3); + + key = 7; + i64SetDefault(&i64Obj, key); + pRaw = i64Encode(&i64Obj); + sdbSetRawStatus(pRaw, SDB_STATUS_READY); + ASSERT_EQ(sdbWrite(pSdb, pRaw), 0); + pI64Obj = (SI64Obj *)sdbAcquire(pSdb, SDB_CONSUMER, &key); + ASSERT_NE(pI64Obj, nullptr); + ASSERT_EQ(pI64Obj->key, key); + ASSERT_EQ(pI64Obj->v8, 7); + ASSERT_EQ(pI64Obj->v16, 7); + ASSERT_EQ(pI64Obj->v32, 7000); + ASSERT_EQ(pI64Obj->v64, 7000); + sdbRelease(pSdb, pI64Obj); + + ASSERT_EQ(sdbGetSize(pSdb, SDB_CONSUMER), 1); + ASSERT_EQ(sdbGetTableVer(pSdb, SDB_CONSUMER), 4); + ASSERT_EQ(sdbGetMaxId(pSdb, SDB_CONSUMER), -1); + ASSERT_EQ(mnode.insertTimes, 7); + ASSERT_EQ(mnode.deleteTimes, 3); + } // write version - - // sdb Write ver - - // sdbRead -#endif + ASSERT_EQ(sdbUpdateVer(pSdb, 1), 0); + ASSERT_EQ(sdbUpdateVer(pSdb, 1), 1); ASSERT_EQ(sdbWriteFile(pSdb), 0); + ASSERT_EQ(sdbWriteFile(pSdb), 0); + sdbCleanup(pSdb); + ASSERT_EQ(mnode.insertTimes, 7); + ASSERT_EQ(mnode.deleteTimes, 7); } -TEST_F(MndTestSdb, 01_Read) { - void *pIter; - int32_t num; - SStrObj *pObj; - SMnode mnode; - SSdb *pSdb; +TEST_F(MndTestSdb, 01_Read_Str) { + void *pIter = NULL; + int32_t num = 0; + SStrObj *pObj = NULL; + SMnode mnode = {0}; + SSdb *pSdb = NULL; SSdbOpt opt = {0}; + SStrObj strObj = {0}; + SSdbRaw *pRaw = NULL; int32_t p1 = 0; int32_t p2 = 111; int32_t p3 = 222; @@ -301,22 +731,174 @@ TEST_F(MndTestSdb, 01_Read) { mnode.v200 = 200; opt.pMnode = &mnode; opt.path = "/tmp/mnode_test_sdb"; - taosRemoveDir(opt.path); - SSdbTable strTable = { - SDB_USER, - SDB_KEY_BINARY, - (SdbDeployFp)strDefault, - (SdbEncodeFp)strEncode, - (SdbDecodeFp)strDecode, - (SdbInsertFp)strInsert, - (SdbUpdateFp)strUpdate, - (SdbDeleteFp)strDelete, - }; + SSdbTable strTable1; + memset(&strTable1, 0, sizeof(SSdbTable)); + strTable1.sdbType = SDB_USER; + strTable1.keyType = SDB_KEY_BINARY; + strTable1.deployFp = (SdbDeployFp)strDefault; + strTable1.encodeFp = (SdbEncodeFp)strEncode; + strTable1.decodeFp = (SdbDecodeFp)strDecode; + strTable1.insertFp = (SdbInsertFp)strInsert; + strTable1.updateFp = (SdbUpdateFp)strUpdate; + strTable1.deleteFp = (SdbDeleteFp)strDelete; + + SSdbTable strTable2; + memset(&strTable2, 0, sizeof(SSdbTable)); + strTable2.sdbType = SDB_VGROUP; + strTable2.keyType = SDB_KEY_INT32; + strTable2.encodeFp = (SdbEncodeFp)i32Encode; + strTable2.decodeFp = (SdbDecodeFp)i32Decode; + strTable2.insertFp = (SdbInsertFp)i32Insert; + strTable2.updateFp = (SdbUpdateFp)i32Update; + strTable2.deleteFp = (SdbDeleteFp)i32Delete; + + SSdbTable strTable3; + memset(&strTable3, 0, sizeof(SSdbTable)); + strTable3.sdbType = SDB_CONSUMER; + strTable3.keyType = SDB_KEY_INT64; + strTable3.encodeFp = (SdbEncodeFp)i64Encode; + strTable3.decodeFp = (SdbDecodeFp)i64Decode; + strTable3.insertFp = (SdbInsertFp)i64Insert; + strTable3.updateFp = (SdbUpdateFp)i64Update; + strTable3.deleteFp = (SdbDeleteFp)i64Delete; pSdb = sdbInit(&opt); mnode.pSdb = pSdb; - + ASSERT_NE(pSdb, nullptr); + ASSERT_NE(pSdb, nullptr); + ASSERT_EQ(sdbSetTable(pSdb, strTable1), 0); + ASSERT_EQ(sdbSetTable(pSdb, strTable2), 0); + ASSERT_EQ(sdbSetTable(pSdb, strTable3), 0); ASSERT_EQ(sdbReadFile(pSdb), 0); + + ASSERT_EQ(sdbGetSize(pSdb, SDB_USER), 2); + ASSERT_EQ(sdbGetMaxId(pSdb, SDB_USER), -1); + ASSERT_EQ(sdbGetTableVer(pSdb, SDB_USER), 5); + ASSERT_EQ(sdbUpdateVer(pSdb, 0), 1); + ASSERT_EQ(mnode.insertTimes, 4); + ASSERT_EQ(mnode.deleteTimes, 0); + + pObj = (SStrObj *)sdbAcquire(pSdb, SDB_USER, "k1000"); + ASSERT_NE(pObj, nullptr); + EXPECT_STREQ(pObj->key, "k1000"); + EXPECT_STREQ(pObj->vstr, "v1000"); + ASSERT_EQ(pObj->v8, 1); + ASSERT_EQ(pObj->v16, 1); + ASSERT_EQ(pObj->v32, 1000); + ASSERT_EQ(pObj->v64, 1000); + sdbRelease(pSdb, pObj); + + pObj = (SStrObj *)sdbAcquire(pSdb, SDB_USER, "k2000"); + ASSERT_NE(pObj, nullptr); + EXPECT_STREQ(pObj->key, "k2000"); + EXPECT_STREQ(pObj->vstr, "v2000"); + ASSERT_EQ(pObj->v8, 2); + ASSERT_EQ(pObj->v16, 2); + ASSERT_EQ(pObj->v32, 2000); + ASSERT_EQ(pObj->v64, 2000); + sdbRelease(pSdb, pObj); + + pObj = (SStrObj *)sdbAcquire(pSdb, SDB_USER, "k200"); + ASSERT_EQ(pObj, nullptr); + + pIter = NULL; + num = 0; + do { + pIter = sdbFetch(pSdb, SDB_USER, pIter, (void **)&pObj); + if (pIter == NULL) break; + ASSERT_NE(pObj, nullptr); + num++; + sdbRelease(pSdb, pObj); + } while (1); + ASSERT_EQ(num, 2); + + do { + pIter = sdbFetch(pSdb, SDB_USER, pIter, (void **)&pObj); + if (pIter == NULL) break; + if (strcmp(pObj->key, "k1000") == 0) { + sdbCancelFetch(pSdb, pIter); + break; + } + } while (1); + EXPECT_STREQ(pObj->key, "k1000"); + + p1 = 0; + p2 = 111; + p3 = 222; + sdbTraverse(pSdb, SDB_USER, (sdbTraverseFp)sdbTraverseSucc1, &p1, &p2, &p3); + ASSERT_EQ(p1, 334); + + p1 = 0; + p2 = 111; + p3 = 222; + sdbTraverse(pSdb, SDB_USER, (sdbTraverseFp)sdbTraverseSucc2, &p1, &p2, &p3); + ASSERT_EQ(p1, 669); + + p1 = 0; + p2 = 111; + p3 = 222; + sdbTraverse(pSdb, SDB_USER, (sdbTraverseFp)sdbTraverseFail, &p1, &p2, &p3); + ASSERT_EQ(p1, 333); + + int32_t i32key = 6; + SI32Obj *pI32Obj = (SI32Obj *)sdbAcquire(pSdb, SDB_VGROUP, &i32key); + ASSERT_NE(pI32Obj, nullptr); + ASSERT_EQ(pI32Obj->key, 6); + ASSERT_EQ(pI32Obj->v8, 6); + ASSERT_EQ(pI32Obj->v16, 6); + ASSERT_EQ(pI32Obj->v32, 6000); + ASSERT_EQ(pI32Obj->v64, 6000); + sdbRelease(pSdb, pI32Obj); + + ASSERT_EQ(sdbGetSize(pSdb, SDB_VGROUP), 1); + ASSERT_EQ(sdbGetTableVer(pSdb, SDB_VGROUP), 4); + ASSERT_EQ(sdbGetMaxId(pSdb, SDB_VGROUP), 7); + + int64_t i64key = 7; + SI64Obj *pI64Obj = (SI64Obj *)sdbAcquire(pSdb, SDB_CONSUMER, &i64key); + ASSERT_NE(pI64Obj, nullptr); + ASSERT_EQ(pI64Obj->key, 7); + ASSERT_EQ(pI64Obj->v8, 7); + ASSERT_EQ(pI64Obj->v16, 7); + ASSERT_EQ(pI64Obj->v32, 7000); + ASSERT_EQ(pI64Obj->v64, 7000); + sdbRelease(pSdb, pI64Obj); + + ASSERT_EQ(sdbGetSize(pSdb, SDB_CONSUMER), 1); + ASSERT_EQ(sdbGetTableVer(pSdb, SDB_CONSUMER), 4); + + ASSERT_EQ(mnode.insertTimes, 4); + ASSERT_EQ(mnode.deleteTimes, 0); + + { + SI32Obj i32Obj = {0}; + int32_t key = 6; + i32SetDefault(&i32Obj, key); + pRaw = i32Encode(&i32Obj); + sdbSetRawStatus(pRaw, SDB_STATUS_DROPPING); + ASSERT_EQ(sdbWrite(pSdb, pRaw), 0); + pI32Obj = (SI32Obj *)sdbAcquire(pSdb, SDB_VGROUP, &key); + ASSERT_EQ(pI32Obj, nullptr); + int32_t code = terrno; + ASSERT_EQ(code, TSDB_CODE_SDB_OBJ_DROPPING); + } + + { + SI32Obj i32Obj = {0}; + int32_t key = 8; + i32SetDefault(&i32Obj, key); + pRaw = i32Encode(&i32Obj); + EXPECT_NE(sdbSetRawStatus(pRaw, SDB_STATUS_INIT), 0); + sdbSetRawStatus(pRaw, SDB_STATUS_CREATING); + ASSERT_EQ(sdbWrite(pSdb, pRaw), 0); + pI32Obj = (SI32Obj *)sdbAcquire(pSdb, SDB_VGROUP, &key); + ASSERT_EQ(pI32Obj, nullptr); + int32_t code = terrno; + ASSERT_EQ(code, TSDB_CODE_SDB_OBJ_CREATING); + } + sdbCleanup(pSdb); + ASSERT_EQ(mnode.insertTimes, 5); + ASSERT_EQ(mnode.deleteTimes, 5); } \ No newline at end of file diff --git a/source/dnode/mnode/sdb/inc/sdbInt.h b/source/dnode/mnode/sdb/inc/sdbInt.h index 563fc72d00..23c0f8a01c 100644 --- a/source/dnode/mnode/sdb/inc/sdbInt.h +++ b/source/dnode/mnode/sdb/inc/sdbInt.h @@ -31,8 +31,6 @@ extern "C" { #define mDebug(...) { if (mDebugFlag & DEBUG_DEBUG) { taosPrintLog("MND ", DEBUG_DEBUG, mDebugFlag, __VA_ARGS__); }} #define mTrace(...) { if (mDebugFlag & DEBUG_TRACE) { taosPrintLog("MND ", DEBUG_TRACE, mDebugFlag, __VA_ARGS__); }} -#define SDB_MAX_SIZE (32 * 1024) - typedef struct SSdbRaw { int8_t type; int8_t status; diff --git a/source/dnode/mnode/sdb/src/sdb.c b/source/dnode/mnode/sdb/src/sdb.c index d180271175..51f40c12cd 100644 --- a/source/dnode/mnode/sdb/src/sdb.c +++ b/source/dnode/mnode/sdb/src/sdb.c @@ -28,12 +28,12 @@ SSdb *sdbInit(SSdbOpt *pOption) { return NULL; } - char path[PATH_MAX + 100]; - snprintf(path, PATH_MAX + 100, "%s%sdata", pOption->path, TD_DIRSEP); + char path[PATH_MAX + 100] = {0}; + snprintf(path, sizeof(path), "%s%sdata", pOption->path, TD_DIRSEP); pSdb->currDir = strdup(path); - snprintf(path, PATH_MAX + 100, "%s%ssync", pOption->path, TD_DIRSEP); + snprintf(path, sizeof(path), "%s%ssync", pOption->path, TD_DIRSEP); pSdb->syncDir = strdup(path); - snprintf(path, PATH_MAX + 100, "%s%stmp", pOption->path, TD_DIRSEP); + snprintf(path, sizeof(path), "%s%stmp", pOption->path, TD_DIRSEP); pSdb->tmpDir = strdup(path); if (pSdb->currDir == NULL || pSdb->currDir == NULL || pSdb->currDir == NULL) { sdbCleanup(pSdb); @@ -50,7 +50,7 @@ SSdb *sdbInit(SSdbOpt *pOption) { for (ESdbType i = 0; i < SDB_MAX; ++i) { taosInitRWLatch(&pSdb->locks[i]); pSdb->maxId[i] = 0; - pSdb->tableVer[i] = -1; + pSdb->tableVer[i] = 0; pSdb->keyTypes[i] = SDB_KEY_INT32; } @@ -99,7 +99,7 @@ void sdbCleanup(SSdb *pSdb) { taosHashClear(hash); taosHashCleanup(hash); pSdb->hashObjs[i] = NULL; - mDebug("sdb table:%d is cleaned up", i); + mDebug("sdb table:%s is cleaned up", sdbTableName(i)); } taosMemoryFree(pSdb); diff --git a/source/dnode/mnode/sdb/src/sdbFile.c b/source/dnode/mnode/sdb/src/sdbFile.c index 33383802be..f8bd14813c 100644 --- a/source/dnode/mnode/sdb/src/sdbFile.c +++ b/source/dnode/mnode/sdb/src/sdbFile.c @@ -18,7 +18,7 @@ #include "tchecksum.h" #include "wal.h" -#define SDB_TABLE_SIZE 24 +#define SDB_TABLE_SIZE 24 #define SDB_RESERVE_SIZE 512 static int32_t sdbRunDeployFp(SSdb *pSdb) { @@ -50,7 +50,7 @@ static int32_t sdbReadFileHead(SSdb *pSdb, TdFilePtr pFile) { } for (int32_t i = 0; i < SDB_TABLE_SIZE; ++i) { - int64_t maxId = -1; + int64_t maxId = 0; ret = taosReadFile(pFile, &maxId, sizeof(int64_t)); if (ret < 0) { terrno = TAOS_SYSTEM_ERROR(errno); @@ -66,7 +66,7 @@ static int32_t sdbReadFileHead(SSdb *pSdb, TdFilePtr pFile) { } for (int32_t i = 0; i < SDB_TABLE_SIZE; ++i) { - int64_t ver = -1; + int64_t ver = 0; ret = taosReadFile(pFile, &ver, sizeof(int64_t)); if (ret < 0) { terrno = TAOS_SYSTEM_ERROR(errno); @@ -102,7 +102,7 @@ static int32_t sdbWriteFileHead(SSdb *pSdb, TdFilePtr pFile) { } for (int32_t i = 0; i < SDB_TABLE_SIZE; ++i) { - int64_t maxId = -1; + int64_t maxId = 0; if (i < SDB_MAX) { maxId = pSdb->maxId[i]; } @@ -113,7 +113,7 @@ static int32_t sdbWriteFileHead(SSdb *pSdb, TdFilePtr pFile) { } for (int32_t i = 0; i < SDB_TABLE_SIZE; ++i) { - int64_t ver = -1; + int64_t ver = 0; if (i < SDB_MAX) { ver = pSdb->tableVer[i]; } @@ -165,6 +165,9 @@ int32_t sdbReadFile(SSdb *pSdb) { return -1; } + int64_t tableVer[SDB_MAX] = {0}; + memcpy(tableVer, pSdb->tableVer, sizeof(tableVer)); + while (1) { readLen = sizeof(SSdbRaw); ret = taosReadFile(pFile, pRaw, readLen); @@ -206,15 +209,16 @@ int32_t sdbReadFile(SSdb *pSdb) { code = sdbWriteWithoutFree(pSdb, pRaw); if (code != 0) { mError("failed to read file:%s since %s", file, terrstr()); - goto PARSE_SDB_DATA_ERROR; + goto _OVER; } } code = 0; pSdb->lastCommitVer = pSdb->curVer; + memcpy(pSdb->tableVer, tableVer, sizeof(tableVer)); mDebug("read file:%s successfully, ver:%" PRId64, file, pSdb->lastCommitVer); -PARSE_SDB_DATA_ERROR: +_OVER: taosCloseFile(&pFile); sdbFreeRaw(pRaw); @@ -259,7 +263,13 @@ static int32_t sdbWriteFileImp(SSdb *pSdb) { SSdbRow **ppRow = taosHashIterate(hash, NULL); while (ppRow != NULL) { SSdbRow *pRow = *ppRow; - if (pRow == NULL || pRow->status != SDB_STATUS_READY) { + if (pRow == NULL) { + ppRow = taosHashIterate(hash, ppRow); + continue; + } + + if (pRow->status != SDB_STATUS_READY && pRow->status != SDB_STATUS_DROPPING) { + sdbPrintOper(pSdb, pRow, "not-write"); ppRow = taosHashIterate(hash, ppRow); continue; } diff --git a/source/dnode/mnode/sdb/src/sdbHash.c b/source/dnode/mnode/sdb/src/sdbHash.c index b34c2a82d8..94008b2f7c 100644 --- a/source/dnode/mnode/sdb/src/sdbHash.c +++ b/source/dnode/mnode/sdb/src/sdbHash.c @@ -16,7 +16,7 @@ #define _DEFAULT_SOURCE #include "sdbInt.h" -static void sdbCheck(SSdb *pSdb, SSdbRow *pRow); +static void sdbCheckRow(SSdb *pSdb, SSdbRow *pRow); const char *sdbTableName(ESdbType type) { switch (type) { @@ -65,12 +65,10 @@ const char *sdbTableName(ESdbType type) { } } -static const char *sdbStatusStr(ESdbStatus status) { +static const char *sdbStatusName(ESdbStatus status) { switch (status) { case SDB_STATUS_CREATING: return "creating"; - case SDB_STATUS_UPDATING: - return "updating"; case SDB_STATUS_DROPPING: return "dropping"; case SDB_STATUS_READY: @@ -89,13 +87,13 @@ void sdbPrintOper(SSdb *pSdb, SSdbRow *pRow, const char *oper) { if (keyType == SDB_KEY_BINARY) { mTrace("%s:%s, ref:%d oper:%s row:%p status:%s", sdbTableName(pRow->type), (char *)pRow->pObj, pRow->refCount, oper, - pRow->pObj, sdbStatusStr(pRow->status)); + pRow->pObj, sdbStatusName(pRow->status)); } else if (keyType == SDB_KEY_INT32) { mTrace("%s:%d, ref:%d oper:%s row:%p status:%s", sdbTableName(pRow->type), *(int32_t *)pRow->pObj, pRow->refCount, - oper, pRow->pObj, sdbStatusStr(pRow->status)); + oper, pRow->pObj, sdbStatusName(pRow->status)); } else if (keyType == SDB_KEY_INT64) { mTrace("%s:%" PRId64 ", ref:%d oper:%s row:%p status:%s", sdbTableName(pRow->type), *(int64_t *)pRow->pObj, - pRow->refCount, oper, pRow->pObj, sdbStatusStr(pRow->status)); + pRow->refCount, oper, pRow->pObj, sdbStatusName(pRow->status)); } else { } } @@ -116,7 +114,7 @@ static SHashObj *sdbGetHash(SSdb *pSdb, int32_t type) { } static int32_t sdbGetkeySize(SSdb *pSdb, ESdbType type, const void *pKey) { - int32_t keySize; + int32_t keySize = 0; EKeyType keyType = pSdb->keyTypes[type]; if (keyType == SDB_KEY_INT32) { @@ -149,7 +147,7 @@ static int32_t sdbInsertRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow * if (taosHashPut(hash, pRow->pObj, keySize, &pRow, sizeof(void *)) != 0) { taosWUnLockLatch(pLock); sdbFreeRow(pSdb, pRow, false); - terrno = TSDB_CODE_SDB_OBJ_ALREADY_THERE; + terrno = TSDB_CODE_OUT_OF_MEMORY; return terrno; } @@ -183,18 +181,18 @@ static int32_t sdbInsertRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow * static int32_t sdbUpdateRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *pNewRow, int32_t keySize) { SRWLatch *pLock = &pSdb->locks[pNewRow->type]; - taosRLockLatch(pLock); + taosWLockLatch(pLock); SSdbRow **ppOldRow = taosHashGet(hash, pNewRow->pObj, keySize); if (ppOldRow == NULL || *ppOldRow == NULL) { - taosRUnLockLatch(pLock); + taosWUnLockLatch(pLock); return sdbInsertRow(pSdb, hash, pRaw, pNewRow, keySize); } SSdbRow *pOldRow = *ppOldRow; pOldRow->status = pRaw->status; sdbPrintOper(pSdb, pOldRow, "update"); - taosRUnLockLatch(pLock); + taosWUnLockLatch(pLock); int32_t code = 0; SdbUpdateFp updateFp = pSdb->updateFps[pNewRow->type]; @@ -230,8 +228,7 @@ static int32_t sdbDeleteRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow * pSdb->tableVer[pOldRow->type]++; sdbFreeRow(pSdb, pRow, false); - sdbCheck(pSdb, pOldRow); - // sdbRelease(pSdb, pOldRow->pObj); + sdbCheckRow(pSdb, pOldRow); return 0; } @@ -254,7 +251,6 @@ int32_t sdbWriteWithoutFree(SSdb *pSdb, SSdbRaw *pRaw) { case SDB_STATUS_CREATING: code = sdbInsertRow(pSdb, hash, pRaw, pRow, keySize); break; - case SDB_STATUS_UPDATING: case SDB_STATUS_READY: case SDB_STATUS_DROPPING: code = sdbUpdateRow(pSdb, hash, pRaw, pRow, keySize); @@ -295,7 +291,6 @@ void *sdbAcquire(SSdb *pSdb, ESdbType type, const void *pKey) { SSdbRow *pRow = *ppRow; switch (pRow->status) { case SDB_STATUS_READY: - case SDB_STATUS_UPDATING: atomic_add_fetch_32(&pRow->refCount, 1); pRet = pRow->pObj; sdbPrintOper(pSdb, pRow, "acquire"); @@ -315,9 +310,9 @@ void *sdbAcquire(SSdb *pSdb, ESdbType type, const void *pKey) { return pRet; } -static void sdbCheck(SSdb *pSdb, SSdbRow *pRow) { +static void sdbCheckRow(SSdb *pSdb, SSdbRow *pRow) { SRWLatch *pLock = &pSdb->locks[pRow->type]; - taosRLockLatch(pLock); + taosWLockLatch(pLock); int32_t ref = atomic_load_32(&pRow->refCount); sdbPrintOper(pSdb, pRow, "check"); @@ -325,7 +320,7 @@ static void sdbCheck(SSdb *pSdb, SSdbRow *pRow) { sdbFreeRow(pSdb, pRow, true); } - taosRUnLockLatch(pLock); + taosWUnLockLatch(pLock); } void sdbRelease(SSdb *pSdb, void *pObj) { @@ -335,7 +330,7 @@ void sdbRelease(SSdb *pSdb, void *pObj) { if (pRow->type >= SDB_MAX) return; SRWLatch *pLock = &pSdb->locks[pRow->type]; - taosRLockLatch(pLock); + taosWLockLatch(pLock); int32_t ref = atomic_sub_fetch_32(&pRow->refCount, 1); sdbPrintOper(pSdb, pRow, "release"); @@ -343,7 +338,7 @@ void sdbRelease(SSdb *pSdb, void *pObj) { sdbFreeRow(pSdb, pRow, true); } - taosRUnLockLatch(pLock); + taosWUnLockLatch(pLock); } void *sdbFetch(SSdb *pSdb, ESdbType type, void *pIter, void **ppObj) { @@ -355,16 +350,6 @@ void *sdbFetch(SSdb *pSdb, ESdbType type, void *pIter, void **ppObj) { SRWLatch *pLock = &pSdb->locks[type]; taosRLockLatch(pLock); -#if 0 - if (pIter != NULL) { - SSdbRow *pLastRow = *(SSdbRow **)pIter; - int32_t ref = atomic_load_32(&pLastRow->refCount); - if (ref <= 0 && pLastRow->status == SDB_STATUS_DROPPED) { - sdbFreeRow(pSdb, pLastRow); - } - } -#endif - SSdbRow **ppRow = taosHashIterate(hash, pIter); while (ppRow != NULL) { SSdbRow *pRow = *ppRow; diff --git a/source/dnode/mnode/sdb/src/sdbRaw.c b/source/dnode/mnode/sdb/src/sdbRaw.c index 45464c3bd3..d09198f66f 100644 --- a/source/dnode/mnode/sdb/src/sdbRaw.c +++ b/source/dnode/mnode/sdb/src/sdbRaw.c @@ -134,6 +134,11 @@ int32_t sdbSetRawStatus(SSdbRaw *pRaw, ESdbStatus status) { return -1; } + if (status == SDB_STATUS_INIT) { + terrno = TSDB_CODE_INVALID_PARA; + return -1; + } + pRaw->status = status; return 0; } diff --git a/source/dnode/vnode/src/vnd/vnodeCommit.c b/source/dnode/vnode/src/vnd/vnodeCommit.c index 5412b5d1c2..7026673231 100644 --- a/source/dnode/vnode/src/vnd/vnodeCommit.c +++ b/source/dnode/vnode/src/vnd/vnodeCommit.c @@ -190,7 +190,7 @@ int vnodeSyncCommit(SVnode *pVnode) { } int vnodeCommit(SVnode *pVnode) { - SVnodeInfo info; + SVnodeInfo info = {0}; char dir[TSDB_FILENAME_LEN]; vInfo("vgId:%d start to commit, version: %" PRId64, TD_VID(pVnode), pVnode->state.applied); diff --git a/source/libs/transport/src/transSrv.c b/source/libs/transport/src/transSrv.c index b416b404a9..916bcb8862 100644 --- a/source/libs/transport/src/transSrv.c +++ b/source/libs/transport/src/transSrv.c @@ -207,7 +207,7 @@ static bool addHandleToAcceptloop(void* arg); SExHandle* exh2 = uvAcquireExHandle(refId); \ if (exh2 == NULL || refId != exh2->refId) { \ tTrace("server handle %p except, may already freed, ignore msg, ref1: %" PRIu64 ", ref2 : %" PRIu64 "", exh1, \ - exh1->refId, refId); \ + exh2 ? exh2->refId : 0, refId); \ goto _return1; \ } \ } else if (refId == 0) { \ diff --git a/source/util/src/terror.c b/source/util/src/terror.c index 87ef94f9b5..709efa3125 100644 --- a/source/util/src/terror.c +++ b/source/util/src/terror.c @@ -93,6 +93,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_OPS_NOT_SUPPORT, "Operation not support TAOS_DEFINE_ERROR(TSDB_CODE_MSG_NOT_PROCESSED, "Message not processed") TAOS_DEFINE_ERROR(TSDB_CODE_CFG_NOT_FOUND, "Config not found") TAOS_DEFINE_ERROR(TSDB_CODE_REPEAT_INIT, "Repeat initialization") +TAOS_DEFINE_ERROR(TSDB_CODE_DUP_KEY, "Cannot add duplicate keys to hash") TAOS_DEFINE_ERROR(TSDB_CODE_REF_NO_MEMORY, "Ref out of memory") TAOS_DEFINE_ERROR(TSDB_CODE_REF_FULL, "too many Ref Objs") diff --git a/source/util/src/thash.c b/source/util/src/thash.c index d2b07875a3..4bbf6ccf45 100644 --- a/source/util/src/thash.c +++ b/source/util/src/thash.c @@ -310,6 +310,7 @@ int32_t taosHashGetSize(const SHashObj *pHashObj) { int32_t taosHashPut(SHashObj *pHashObj, const void *key, size_t keyLen, const void *data, size_t size) { if (pHashObj == NULL || key == NULL || keyLen == 0) { + terrno = TSDB_CODE_INVALID_PTR; return -1; } @@ -378,6 +379,8 @@ int32_t taosHashPut(SHashObj *pHashObj, const void *key, size_t keyLen, const vo } doUpdateHashNode(pHashObj, pe, prev, pNode, pNewNode); + } else { + terrno = TSDB_CODE_DUP_KEY; } taosHashEntryWUnlock(pHashObj, pe); diff --git a/source/util/src/tprocess.c b/source/util/src/tprocess.c index d8343bc427..8963a4f94e 100644 --- a/source/util/src/tprocess.c +++ b/source/util/src/tprocess.c @@ -175,7 +175,6 @@ static int32_t taosProcQueuePush(SProcObj *pProc, SProcQueue *pQueue, const char if (handle != 0 && ftype == PROC_FUNC_REQ) { if (taosHashPut(pProc->hash, &handle, sizeof(int64_t), &handleRef, sizeof(int64_t)) != 0) { taosThreadMutexUnlock(&pQueue->mutex); - terrno = TSDB_CODE_OUT_OF_MEMORY; return -1; } } @@ -227,8 +226,8 @@ static int32_t taosProcQueuePush(SProcObj *pProc, SProcQueue *pQueue, const char taosThreadMutexUnlock(&pQueue->mutex); tsem_post(&pQueue->sem); - uTrace("proc:%s, push msg at pos:%d ftype:%d remain:%d, head:%d %p body:%d %p", pQueue->name, pos, ftype, - pQueue->items, headLen, pHead, bodyLen, pBody); + uTrace("proc:%s, push msg at pos:%d ftype:%d remain:%d handle:%p ref:%" PRId64 ", head:%d %p body:%d %p", + pQueue->name, pos, ftype, pQueue->items, (void *)handle, handleRef, headLen, pHead, bodyLen, pBody); return 0; } @@ -455,25 +454,28 @@ void taosProcCleanup(SProcObj *pProc) { } int32_t taosProcPutToChildQ(SProcObj *pProc, const void *pHead, int16_t headLen, const void *pBody, int32_t bodyLen, - void *handle, int64_t handleRef, EProcFuncType ftype) { + void *handle, int64_t ref, EProcFuncType ftype) { if (ftype != PROC_FUNC_REQ) { terrno = TSDB_CODE_INVALID_PARA; return -1; } - return taosProcQueuePush(pProc, pProc->pChildQueue, pHead, headLen, pBody, bodyLen, (int64_t)handle, handleRef, - ftype); + return taosProcQueuePush(pProc, pProc->pChildQueue, pHead, headLen, pBody, bodyLen, (int64_t)handle, ref, ftype); } int64_t taosProcRemoveHandle(SProcObj *pProc, void *handle) { int64_t h = (int64_t)handle; taosThreadMutexLock(&pProc->pChildQueue->mutex); - int64_t *handleRef = taosHashGet(pProc->hash, &h, sizeof(int64_t)); + int64_t *pRef = taosHashGet(pProc->hash, &h, sizeof(int64_t)); + int64_t ref = 0; + if (pRef != NULL) { + ref = *pRef; + } + taosHashRemove(pProc->hash, &h, sizeof(int64_t)); taosThreadMutexUnlock(&pProc->pChildQueue->mutex); - if (handleRef == NULL) return 0; - return *handleRef; + return ref; } void taosProcCloseHandles(SProcObj *pProc, void (*HandleFp)(void *handle)) { diff --git a/tests/script/jenkins/basic.txt b/tests/script/jenkins/basic.txt index 6f91235d90..d1c5e83975 100644 --- a/tests/script/jenkins/basic.txt +++ b/tests/script/jenkins/basic.txt @@ -66,10 +66,10 @@ # --- stable ./test.sh -f tsim/stable/disk.sim -#./test.sh -f tsim/stable/dnode3.sim +./test.sh -f tsim/stable/dnode3.sim ./test.sh -f tsim/stable/metrics.sim ./test.sh -f tsim/stable/refcount.sim -# ./test.sh -f tsim/stable/show.sim +#./test.sh -f tsim/stable/show.sim ./test.sh -f tsim/stable/values.sim ./test.sh -f tsim/stable/vnode3.sim @@ -81,7 +81,7 @@ ./test.sh -f tsim/insert/backquote.sim -m ./test.sh -f tsim/parser/fourArithmetic-basic.sim -m ./test.sh -f tsim/query/interval-offset.sim -m -#./test.sh -f tsim/tmq/basic3.sim -m +./test.sh -f tsim/tmq/basic3.sim -m ./test.sh -f tsim/stable/vnode3.sim -m ./test.sh -f tsim/qnode/basic1.sim -m ./test.sh -f tsim/mnode/basic1.sim -m diff --git a/tests/script/tsim/show/basic.sim b/tests/script/tsim/show/basic.sim index abf5733a95..e171d1abb9 100644 --- a/tests/script/tsim/show/basic.sim +++ b/tests/script/tsim/show/basic.sim @@ -3,9 +3,6 @@ system sh/deploy.sh -n dnode1 -i 1 system sh/deploy.sh -n dnode2 -i 2 system sh/exec.sh -n dnode1 -s start system sh/exec.sh -n dnode2 -s start - -# after mnode support, del sleep 2000, and check dnode status -sleep 2000 sql connect #$loop_cnt = 0